From afab2efccd33462f4bcdf5e3a2894e543da98f9a Mon Sep 17 00:00:00 2001 From: 4pmtong Date: Fri, 7 Aug 2026 16:05:05 +0800 Subject: [PATCH] feat: add workforce bundle environment contracts --- backend/app/run_journal/__init__.py | 6 + backend/app/run_journal/models.py | 51 ++ backend/app/run_journal/store.py | 594 ++++++++++++++- backend/app/workspace_config/__init__.py | 61 ++ backend/app/workspace_config/manifest.py | 51 ++ backend/app/workspace_config/models.py | 708 ++++++++++++++++++ backend/app/workspace_config/resolver.py | 81 ++ backend/pyproject.toml | 1 + backend/tests/app/run_journal/test_store.py | 189 +++++ .../tests/app/workspace_config/__init__.py | 1 + .../tests/app/workspace_config/test_models.py | 246 ++++++ backend/uv.lock | 6 +- 12 files changed, 1989 insertions(+), 6 deletions(-) create mode 100644 backend/app/workspace_config/__init__.py create mode 100644 backend/app/workspace_config/manifest.py create mode 100644 backend/app/workspace_config/models.py create mode 100644 backend/app/workspace_config/resolver.py create mode 100644 backend/tests/app/workspace_config/__init__.py create mode 100644 backend/tests/app/workspace_config/test_models.py diff --git a/backend/app/run_journal/__init__.py b/backend/app/run_journal/__init__.py index 2fdfbedb1..d67cc930b 100644 --- a/backend/app/run_journal/__init__.py +++ b/backend/app/run_journal/__init__.py @@ -14,11 +14,13 @@ from app.run_journal.models import ( ApprovalRecord, + AttemptEnvironmentBinding, CloudRunEventReplica, CloudRunReplica, CommandResultEvent, CommandResultSyncBatch, CommittedRunEvent, + EffectiveEnvironmentSpecRecord, RemoteCommandInboxRecord, RunAttemptRecord, RunEventDraft, @@ -27,6 +29,7 @@ from app.run_journal.models import ( RunRecord, StartupReconciliationResult, ToolCallRecord, + WorkspaceConfigRevisionRecord, ) from app.run_journal.paths import default_run_journal_path from app.run_journal.recorder import EventRecorder @@ -52,12 +55,14 @@ from app.run_journal.store import ( __all__ = [ "SCHEMA_VERSION", "ApprovalRecord", + "AttemptEnvironmentBinding", "CloudRunEventReplica", "CloudRunReplica", "CommittedRunEvent", "CommandResultEvent", "CommandResultSyncBatch", "EventRecorder", + "EffectiveEnvironmentSpecRecord", "IdempotencyConflictError", "InvalidRunTransitionError", "OptimisticConcurrencyError", @@ -73,6 +78,7 @@ __all__ = [ "SQLiteRunJournal", "StartupReconciliationResult", "ToolCallRecord", + "WorkspaceConfigRevisionRecord", "UnsafeResumeError", "UnsupportedSchemaVersionError", "close_default_run_journal", diff --git a/backend/app/run_journal/models.py b/backend/app/run_journal/models.py index 5ed5784e8..f94444611 100644 --- a/backend/app/run_journal/models.py +++ b/backend/app/run_journal/models.py @@ -79,6 +79,57 @@ class RunAttemptRecord: policy_version: str elapsed_active_ms: int last_consumer_heartbeat_at: float | None + environment_spec_id: str | None = None + environment_spec_digest: str | None = None + bundle_revision_id: str | None = None + permission_profile_revision: str | None = None + thinking_effort_requested: str | None = None + thinking_effort_effective: str | None = None + provider_capability_revision: str | None = None + + +@dataclass(frozen=True) +class AttemptEnvironmentBinding: + environment_spec_id: str + environment_spec_digest: str + bundle_revision_id: str + permission_profile_revision: str + thinking_effort_requested: str + thinking_effort_effective: str + provider_capability_revision: str + + +@dataclass(frozen=True) +class WorkspaceConfigRevisionRecord: + revision_id: str + space_id: str + bundle_id: str + revision_number: int + config_placement: str + status: str + version: int + manifest: dict[str, Any] + manifest_digest: str + created_by: str + created_at: float + + +@dataclass(frozen=True) +class EffectiveEnvironmentSpecRecord: + environment_spec_id: str + owner_type: str + owner_id: str + bundle_revision_id: str + manifest_digest: str + spec: dict[str, Any] + environment_spec_digest: str + semantic_spec_digest: str + local_materialization_digest: str + redacted_spec: dict[str, Any] + projection_digest: str + permission_profile_revision: str + provider_capability_revision: str + created_at: float @dataclass(frozen=True) diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index 64cd6e504..5df2f7078 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -35,11 +35,13 @@ from typing import Any from app.run_journal.models import ( ApprovalRecord, + AttemptEnvironmentBinding, CloudRunEventReplica, CloudRunReplica, CommandResultEvent, CommandResultSyncBatch, CommittedRunEvent, + EffectiveEnvironmentSpecRecord, RemoteCommandInboxRecord, RunAttemptRecord, RunEventDraft, @@ -48,6 +50,7 @@ from app.run_journal.models import ( RunRecord, StartupReconciliationResult, ToolCallRecord, + WorkspaceConfigRevisionRecord, ) from app.run_journal.paths import default_run_journal_path from app.run_journal.transitions import ( @@ -65,8 +68,14 @@ from app.run_policy import ( ToolSafetyClass, automatic_tool_replay_allowed, ) +from app.workspace_config.models import ( + EffectiveEnvironmentSpec, + ThinkingEffort, + canonical_digest, + canonical_json, +) -SCHEMA_VERSION = 5 +SCHEMA_VERSION = 6 logger = logging.getLogger("run_journal") _MIGRATION_V1 = """ @@ -330,6 +339,98 @@ PRAGMA user_version = 5; COMMIT; """ +_MIGRATION_V6 = """ +BEGIN IMMEDIATE; + +CREATE TABLE workspace_config_revisions ( + revision_id TEXT PRIMARY KEY, + space_id TEXT NOT NULL, + bundle_id TEXT NOT NULL, + revision_number INTEGER NOT NULL CHECK (revision_number > 0), + config_placement TEXT NOT NULL CHECK ( + config_placement IN ('in_repo', 'sidecar') + ), + status TEXT NOT NULL CHECK ( + status IN ('draft', 'validated', 'published', 'deprecated') + ), + version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0), + manifest_json TEXT NOT NULL, + manifest_digest TEXT NOT NULL CHECK (length(manifest_digest) = 64), + created_by TEXT NOT NULL, + created_at REAL NOT NULL, + UNIQUE(space_id, revision_number) +); + +CREATE INDEX workspace_config_revisions_space_idx +ON workspace_config_revisions(space_id, revision_number DESC); + +CREATE TABLE workspace_config_materializations ( + materialization_id TEXT PRIMARY KEY, + space_id TEXT NOT NULL, + revision_id TEXT NOT NULL REFERENCES workspace_config_revisions( + revision_id + ) ON DELETE RESTRICT, + state TEXT NOT NULL CHECK ( + state IN ('pending', 'materialized', 'needs_attention', 'degraded') + ), + local_override_digest TEXT NOT NULL DEFAULT '', + materialized_at REAL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(space_id, revision_id, local_override_digest) +); + +CREATE TABLE effective_environment_specs ( + environment_spec_id TEXT PRIMARY KEY, + owner_type TEXT NOT NULL CHECK (owner_type IN ('run', 'run_attempt')), + owner_id TEXT NOT NULL, + bundle_revision_id TEXT NOT NULL REFERENCES workspace_config_revisions( + revision_id + ) ON DELETE RESTRICT, + manifest_digest TEXT NOT NULL CHECK (length(manifest_digest) = 64), + spec_json TEXT NOT NULL, + environment_spec_digest TEXT NOT NULL CHECK ( + length(environment_spec_digest) = 64 + ), + semantic_spec_digest TEXT NOT NULL CHECK ( + length(semantic_spec_digest) = 64 + ), + local_materialization_digest TEXT NOT NULL CHECK ( + length(local_materialization_digest) = 64 + ), + redacted_spec_json TEXT NOT NULL, + projection_digest TEXT NOT NULL CHECK (length(projection_digest) = 64), + permission_profile_revision TEXT NOT NULL, + provider_capability_revision TEXT NOT NULL, + created_at REAL NOT NULL +); + +CREATE INDEX effective_environment_specs_owner_idx +ON effective_environment_specs(owner_type, owner_id, created_at DESC); + +ALTER TABLE run_attempts ADD COLUMN environment_spec_id TEXT REFERENCES + effective_environment_specs(environment_spec_id) ON DELETE RESTRICT; +ALTER TABLE run_attempts ADD COLUMN environment_spec_digest TEXT; +ALTER TABLE run_attempts ADD COLUMN bundle_revision_id TEXT REFERENCES + workspace_config_revisions(revision_id) ON DELETE RESTRICT; +ALTER TABLE run_attempts ADD COLUMN permission_profile_revision TEXT; +ALTER TABLE run_attempts ADD COLUMN thinking_effort_requested TEXT CHECK ( + thinking_effort_requested IS NULL OR + thinking_effort_requested IN ('low', 'medium', 'high', 'xhigh', 'max') +); +ALTER TABLE run_attempts ADD COLUMN thinking_effort_effective TEXT CHECK ( + thinking_effort_effective IS NULL OR + thinking_effort_effective IN ('low', 'medium', 'high', 'xhigh', 'max') +); +ALTER TABLE run_attempts ADD COLUMN provider_capability_revision TEXT; + +INSERT OR IGNORE INTO run_journal_migrations(version, applied_at) +VALUES (6, CAST(strftime('%s', 'now') AS REAL)); + +PRAGMA user_version = 6; +COMMIT; +""" + class RunJournalError(RuntimeError): """Base error for local RunJournal operations.""" @@ -429,6 +530,324 @@ class SQLiteRunJournal: ).fetchone()[0], } + def put_workspace_config_revision( + self, + *, + revision_id: str, + space_id: str, + bundle_id: str, + revision_number: int, + config_placement: str, + manifest: dict[str, Any], + status: str = "validated", + created_by: str, + now: float | None = None, + ) -> WorkspaceConfigRevisionRecord: + """Insert one immutable Bundle revision or return its exact replay.""" + + required = { + "revision_id": revision_id, + "space_id": space_id, + "bundle_id": bundle_id, + "created_by": created_by, + } + for field_name, value in required.items(): + if not value.strip(): + raise ValueError(f"{field_name} is required") + if revision_number < 1: + raise ValueError("revision_number must be positive") + if config_placement not in {"in_repo", "sidecar"}: + raise ValueError("invalid config_placement") + if status not in {"draft", "validated", "published", "deprecated"}: + raise ValueError("invalid workspace config revision status") + timestamp = now if now is not None else time.time() + manifest_json = canonical_json(manifest) + manifest_digest = canonical_digest(manifest) + expected = ( + revision_id, + space_id, + bundle_id, + revision_number, + config_placement, + status, + manifest_json, + manifest_digest, + created_by, + ) + with self._write_transaction() as connection: + row = connection.execute( + """ + SELECT * FROM workspace_config_revisions + WHERE revision_id = ? + OR (space_id = ? AND revision_number = ?) + """, + (revision_id, space_id, revision_number), + ).fetchone() + if row is not None: + actual = ( + row["revision_id"], + row["space_id"], + row["bundle_id"], + int(row["revision_number"]), + row["config_placement"], + row["status"], + row["manifest_json"], + row["manifest_digest"], + row["created_by"], + ) + if actual != expected: + raise IdempotencyConflictError( + f"workspace config revision {revision_id!r} conflicts " + "with an existing revision" + ) + return self._workspace_config_revision_from_row(row) + connection.execute( + """ + INSERT INTO workspace_config_revisions( + revision_id, space_id, bundle_id, revision_number, + config_placement, status, manifest_json, manifest_digest, + created_by, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + (*expected, timestamp), + ) + row = connection.execute( + """ + SELECT * FROM workspace_config_revisions + WHERE revision_id = ? + """, + (revision_id,), + ).fetchone() + assert row is not None + return self._workspace_config_revision_from_row(row) + + def get_workspace_config_revision( + self, revision_id: str + ) -> WorkspaceConfigRevisionRecord | None: + with self._lock: + row = self._connection.execute( + """ + SELECT * FROM workspace_config_revisions + WHERE revision_id = ? + """, + (revision_id,), + ).fetchone() + return ( + self._workspace_config_revision_from_row(row) + if row is not None + else None + ) + + def transition_workspace_config_revision( + self, + revision_id: str, + *, + expected_version: int, + status: str, + ) -> WorkspaceConfigRevisionRecord: + """CAS the Bundle lifecycle without making its manifest mutable.""" + + allowed = { + "draft": {"validated"}, + "validated": {"published"}, + "published": {"deprecated"}, + "deprecated": set(), + } + if status not in allowed: + raise ValueError("invalid workspace config revision status") + with self._write_transaction() as connection: + row = connection.execute( + """ + SELECT * FROM workspace_config_revisions + WHERE revision_id = ? + """, + (revision_id,), + ).fetchone() + if row is None: + raise RunNotFoundError( + f"workspace config revision {revision_id!r} does not exist" + ) + if int(row["version"]) != expected_version: + raise OptimisticConcurrencyError( + f"workspace config revision {revision_id!r} expected " + f"version {expected_version}, found {row['version']}" + ) + if row["status"] == status: + return self._workspace_config_revision_from_row(row) + if status not in allowed[row["status"]]: + raise InvalidRunTransitionError( + f"workspace config revision {revision_id!r} cannot move " + f"from {row['status']!r} to {status!r}" + ) + updated = connection.execute( + """ + UPDATE workspace_config_revisions + SET status = ?, version = version + 1 + WHERE revision_id = ? AND version = ? AND status = ? + """, + ( + status, + revision_id, + expected_version, + row["status"], + ), + ) + if updated.rowcount != 1: + raise OptimisticConcurrencyError( + f"workspace config revision {revision_id!r} changed " + "during transition" + ) + row = connection.execute( + """ + SELECT * FROM workspace_config_revisions + WHERE revision_id = ? + """, + (revision_id,), + ).fetchone() + assert row is not None + return self._workspace_config_revision_from_row(row) + + def put_effective_environment_spec( + self, + spec: EffectiveEnvironmentSpec, + *, + now: float | None = None, + ) -> EffectiveEnvironmentSpecRecord: + """Persist immutable local and redacted forms in one transaction.""" + + timestamp = now if now is not None else time.time() + local_payload = spec.local_payload() + if canonical_digest(spec.semantic_spec) != spec.semantic_spec_digest: + raise IdempotencyConflictError( + "semantic EnvironmentSpec digest does not match its payload" + ) + local_materialization_payload = spec.local_materialization.model_dump( + exclude_none=True, + mode="json", + ) + if ( + canonical_digest(local_materialization_payload) + != spec.local_materialization_digest + ): + raise IdempotencyConflictError( + "local materialization digest does not match its payload" + ) + spec_json = canonical_json(local_payload) + environment_spec_digest = spec.digest + redacted_payload = spec.cloud_projection() + projection_digest = str(redacted_payload["projection_digest"]) + projection_body = { + key: value + for key, value in redacted_payload.items() + if key != "projection_digest" + } + if canonical_digest(projection_body) != projection_digest: + raise IdempotencyConflictError( + "Cloud EnvironmentSpec projection digest is invalid" + ) + redacted_spec_json = canonical_json(redacted_payload) + with self._write_transaction() as connection: + revision = connection.execute( + """ + SELECT manifest_digest FROM workspace_config_revisions + WHERE revision_id = ? + """, + (spec.bundle_revision_id,), + ).fetchone() + if revision is None: + raise RunNotFoundError( + f"workspace config revision " + f"{spec.bundle_revision_id!r} does not exist" + ) + if revision["manifest_digest"] != spec.manifest_digest: + raise IdempotencyConflictError( + "EnvironmentSpec manifest digest does not match its " + "workspace config revision" + ) + expected = ( + spec.spec_id, + spec.owner_type, + spec.owner_id, + spec.bundle_revision_id, + spec.manifest_digest, + spec_json, + environment_spec_digest, + spec.semantic_spec_digest, + spec.local_materialization_digest, + redacted_spec_json, + projection_digest, + spec.permission_profile_revision, + spec.provider_capability_revision, + ) + row = connection.execute( + """ + SELECT * FROM effective_environment_specs + WHERE environment_spec_id = ? + """, + (spec.spec_id,), + ).fetchone() + if row is not None: + actual = ( + row["environment_spec_id"], + row["owner_type"], + row["owner_id"], + row["bundle_revision_id"], + row["manifest_digest"], + row["spec_json"], + row["environment_spec_digest"], + row["semantic_spec_digest"], + row["local_materialization_digest"], + row["redacted_spec_json"], + row["projection_digest"], + row["permission_profile_revision"], + row["provider_capability_revision"], + ) + if actual != expected: + raise IdempotencyConflictError( + f"EnvironmentSpec {spec.spec_id!r} conflicts with " + "an existing immutable spec" + ) + return self._effective_environment_spec_from_row(row) + connection.execute( + """ + INSERT INTO effective_environment_specs( + environment_spec_id, owner_type, owner_id, + bundle_revision_id, manifest_digest, spec_json, + environment_spec_digest, semantic_spec_digest, + local_materialization_digest, redacted_spec_json, + projection_digest, permission_profile_revision, + provider_capability_revision, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + (*expected, timestamp), + ) + row = connection.execute( + """ + SELECT * FROM effective_environment_specs + WHERE environment_spec_id = ? + """, + (spec.spec_id,), + ).fetchone() + assert row is not None + return self._effective_environment_spec_from_row(row) + + def get_effective_environment_spec( + self, environment_spec_id: str + ) -> EffectiveEnvironmentSpecRecord | None: + with self._lock: + row = self._connection.execute( + """ + SELECT * FROM effective_environment_specs + WHERE environment_spec_id = ? + """, + (environment_spec_id,), + ).fetchone() + return ( + self._effective_environment_spec_from_row(row) + if row is not None + else None + ) + def ensure_run( self, *, @@ -911,11 +1330,14 @@ class SQLiteRunJournal: reason: str, activate: bool = False, attempt_id: str | None = None, + environment: AttemptEnvironmentBinding | None = None, now: float | None = None, ) -> RunAttemptRecord: if not request_id.strip() or not reason.strip(): raise ValueError("attempt request_id and reason are required") timestamp = now if now is not None else time.time() + identifier = attempt_id or str(uuid.uuid4()) + environment_values = self._attempt_environment_values(environment) with self._write_transaction() as connection: run = connection.execute( "SELECT * FROM runs WHERE run_id = ?", (run_id,) @@ -934,6 +1356,20 @@ class SQLiteRunJournal: raise IdempotencyConflictError( f"attempt request_id {request_id!r} was reused with a different reason" ) + persisted_environment = ( + duplicate["environment_spec_id"], + duplicate["environment_spec_digest"], + duplicate["bundle_revision_id"], + duplicate["permission_profile_revision"], + duplicate["thinking_effort_requested"], + duplicate["thinking_effort_effective"], + duplicate["provider_capability_revision"], + ) + if persisted_environment != environment_values: + raise IdempotencyConflictError( + f"attempt request_id {request_id!r} was reused with " + "a different environment" + ) return self._attempt_from_row(duplicate) if run["origin"] == "cloud_restore": raise InvalidRunTransitionError( @@ -979,6 +1415,43 @@ class SQLiteRunJournal: blockers = self._unsafe_resume_blockers(connection, run_id) if blockers: raise UnsafeResumeError(blockers) + if environment is not None: + spec = connection.execute( + """ + SELECT * FROM effective_environment_specs + WHERE environment_spec_id = ? + """, + (environment.environment_spec_id,), + ).fetchone() + if spec is None: + raise RunNotFoundError( + f"EnvironmentSpec " + f"{environment.environment_spec_id!r} does not exist" + ) + expected_owner_id = ( + run_id if spec["owner_type"] == "run" else identifier + ) + if spec["owner_id"] != expected_owner_id: + raise IdempotencyConflictError( + "EnvironmentSpec belongs to another Run/Attempt" + ) + persisted_spec_values = ( + spec["environment_spec_digest"], + spec["bundle_revision_id"], + spec["permission_profile_revision"], + spec["provider_capability_revision"], + ) + binding_spec_values = ( + environment.environment_spec_digest, + environment.bundle_revision_id, + environment.permission_profile_revision, + environment.provider_capability_revision, + ) + if persisted_spec_values != binding_spec_values: + raise IdempotencyConflictError( + "Attempt environment binding does not match its " + "immutable EnvironmentSpec" + ) number = int( connection.execute( """ @@ -988,7 +1461,6 @@ class SQLiteRunJournal: (run_id,), ).fetchone()[0] ) - identifier = attempt_id or str(uuid.uuid4()) status = "running" if activate else "pending" connection.execute( """ @@ -996,8 +1468,14 @@ class SQLiteRunJournal: attempt_id, run_id, attempt_number, status, started_at, ended_at, outcome, timeout_reason, resume_request_id, resume_reason, policy_version, elapsed_active_ms, - last_consumer_heartbeat_at - ) VALUES (?, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, ?, 0, ?) + last_consumer_heartbeat_at, environment_spec_id, + environment_spec_digest, bundle_revision_id, + permission_profile_revision, thinking_effort_requested, + thinking_effort_effective, provider_capability_revision + ) VALUES ( + ?, ?, ?, ?, ?, NULL, NULL, NULL, ?, ?, ?, 0, ?, + ?, ?, ?, ?, ?, ?, ? + ) """, ( identifier, @@ -1009,8 +1487,32 @@ class SQLiteRunJournal: reason, run["timeout_policy_version"], timestamp if activate else None, + *environment_values, ), ) + environment_payload = ( + { + "environment_spec_id": environment.environment_spec_id, + "environment_spec_digest": ( + environment.environment_spec_digest + ), + "bundle_revision_id": environment.bundle_revision_id, + "permission_profile_revision": ( + environment.permission_profile_revision + ), + "thinking_effort_requested": ( + environment.thinking_effort_requested + ), + "thinking_effort_effective": ( + environment.thinking_effort_effective + ), + "provider_capability_revision": ( + environment.provider_capability_revision + ), + } + if environment is not None + else {} + ) self._append_event_in_transaction( connection, run_id, @@ -1023,6 +1525,7 @@ class SQLiteRunJournal: "reason": reason, "status": status, "policy_version": run["timeout_policy_version"], + **environment_payload, }, created_at=timestamp, ), @@ -3162,6 +3665,8 @@ class SQLiteRunJournal: self._connection.executescript(_MIGRATION_V4) if version < 5: self._connection.executescript(_MIGRATION_V5) + if version < 6: + self._connection.executescript(_MIGRATION_V6) @contextmanager def _write_transaction(self) -> Iterator[sqlite3.Connection]: @@ -3371,6 +3876,87 @@ class SQLiteRunJournal: if row["last_consumer_heartbeat_at"] is not None else None ), + environment_spec_id=row["environment_spec_id"], + environment_spec_digest=row["environment_spec_digest"], + bundle_revision_id=row["bundle_revision_id"], + permission_profile_revision=row["permission_profile_revision"], + thinking_effort_requested=row["thinking_effort_requested"], + thinking_effort_effective=row["thinking_effort_effective"], + provider_capability_revision=row["provider_capability_revision"], + ) + + @staticmethod + def _attempt_environment_values( + environment: AttemptEnvironmentBinding | None, + ) -> tuple[str | None, ...]: + if environment is None: + return (None, None, None, None, None, None, None) + required = ( + environment.environment_spec_id, + environment.environment_spec_digest, + environment.bundle_revision_id, + environment.permission_profile_revision, + environment.provider_capability_revision, + ) + if any(not value.strip() for value in required): + raise ValueError("Attempt environment binding fields are required") + for value in ( + environment.thinking_effort_requested, + environment.thinking_effort_effective, + ): + try: + ThinkingEffort(value) + except ValueError as exc: + raise ValueError( + f"invalid persisted thinking effort {value!r}" + ) from exc + return ( + environment.environment_spec_id, + environment.environment_spec_digest, + environment.bundle_revision_id, + environment.permission_profile_revision, + environment.thinking_effort_requested, + environment.thinking_effort_effective, + environment.provider_capability_revision, + ) + + @staticmethod + def _workspace_config_revision_from_row( + row: sqlite3.Row, + ) -> WorkspaceConfigRevisionRecord: + return WorkspaceConfigRevisionRecord( + revision_id=row["revision_id"], + space_id=row["space_id"], + bundle_id=row["bundle_id"], + revision_number=int(row["revision_number"]), + config_placement=row["config_placement"], + status=row["status"], + version=int(row["version"]), + manifest=json.loads(row["manifest_json"]), + manifest_digest=row["manifest_digest"], + created_by=row["created_by"], + created_at=float(row["created_at"]), + ) + + @staticmethod + def _effective_environment_spec_from_row( + row: sqlite3.Row, + ) -> EffectiveEnvironmentSpecRecord: + return EffectiveEnvironmentSpecRecord( + environment_spec_id=row["environment_spec_id"], + owner_type=row["owner_type"], + owner_id=row["owner_id"], + bundle_revision_id=row["bundle_revision_id"], + manifest_digest=row["manifest_digest"], + spec=json.loads(row["spec_json"]), + environment_spec_digest=row["environment_spec_digest"], + semantic_spec_digest=row["semantic_spec_digest"], + local_materialization_digest=row["local_materialization_digest"], + redacted_spec=json.loads(row["redacted_spec_json"]), + projection_digest=row["projection_digest"], + permission_profile_revision=row["permission_profile_revision"], + provider_capability_revision=row["provider_capability_revision"], + created_at=float(row["created_at"]), ) @staticmethod diff --git a/backend/app/workspace_config/__init__.py b/backend/app/workspace_config/__init__.py new file mode 100644 index 000000000..94053a05a --- /dev/null +++ b/backend/app/workspace_config/__init__.py @@ -0,0 +1,61 @@ +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +from app.workspace_config.manifest import ( + load_workforce_manifest, + parse_workforce_manifest, +) +from app.workspace_config.models import ( + ConfigPlacement, + EffectiveEnvironmentSpec, + EffortResolution, + LocalMaterialization, + ProviderModelCapability, + ResolvedConnectorBinding, + ResolvedContextSource, + SecretValueInManifestError, + ThinkingEffort, + UnsafeCloudProjectionError, + UnsupportedThinkingEffortError, + WorkforceBundleManifest, + WorkspaceConfigError, + WorktreeMaterialization, + canonical_digest, + canonical_json, + normalize_thinking_effort, +) +from app.workspace_config.resolver import EnvironmentConfigResolver + +__all__ = [ + "ConfigPlacement", + "EffectiveEnvironmentSpec", + "EffortResolution", + "EnvironmentConfigResolver", + "LocalMaterialization", + "ProviderModelCapability", + "ResolvedConnectorBinding", + "ResolvedContextSource", + "SecretValueInManifestError", + "ThinkingEffort", + "UnsafeCloudProjectionError", + "UnsupportedThinkingEffortError", + "WorkforceBundleManifest", + "WorkspaceConfigError", + "WorktreeMaterialization", + "canonical_digest", + "canonical_json", + "load_workforce_manifest", + "normalize_thinking_effort", + "parse_workforce_manifest", +] diff --git a/backend/app/workspace_config/manifest.py b/backend/app/workspace_config/manifest.py new file mode 100644 index 000000000..dffd82e60 --- /dev/null +++ b/backend/app/workspace_config/manifest.py @@ -0,0 +1,51 @@ +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +"""Safe parser for shareable Workforce Bundle manifests.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import yaml + +from app.workspace_config.models import ( + WorkforceBundleManifest, + WorkspaceConfigError, + assert_manifest_secret_free, +) + + +def parse_workforce_manifest(value: str | bytes) -> WorkforceBundleManifest: + try: + payload: Any = yaml.safe_load(value) + except yaml.YAMLError as exc: + raise WorkspaceConfigError("invalid Workforce Bundle YAML") from exc + if not isinstance(payload, dict): + raise WorkspaceConfigError( + "Workforce Bundle manifest must be a YAML mapping" + ) + # Run this before Pydantic so callers receive the typed privacy-boundary + # error instead of a generic wrapped ValidationError. + assert_manifest_secret_free(payload) + return WorkforceBundleManifest.model_validate(payload) + + +def load_workforce_manifest(path: Path) -> WorkforceBundleManifest: + if path.name != "workspace.yaml": + raise WorkspaceConfigError( + "Workforce Bundle manifest must be named workspace.yaml" + ) + return parse_workforce_manifest(path.read_text(encoding="utf-8")) diff --git a/backend/app/workspace_config/models.py b/backend/app/workspace_config/models.py new file mode 100644 index 000000000..f12a22bb4 --- /dev/null +++ b/backend/app/workspace_config/models.py @@ -0,0 +1,708 @@ +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +"""Immutable Workforce Bundle and environment materialization contracts.""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from enum import StrEnum +from types import MappingProxyType +from typing import Any, Literal + +from pydantic import ( + BaseModel, + ConfigDict, + Field, + field_validator, + model_validator, +) + + +class WorkspaceConfigError(ValueError): + """Base error for invalid workspace configuration.""" + + +class SecretValueInManifestError(WorkspaceConfigError): + """Raised when a secret-bearing field appears in a shareable manifest.""" + + +class UnsupportedThinkingEffortError(WorkspaceConfigError): + """Raised when a provider cannot honor a requested effort.""" + + +class UnsafeCloudProjectionError(WorkspaceConfigError): + """Raised when a Cloud projection contains device-local identity.""" + + +class ThinkingEffort(StrEnum): + LOW = "low" + MEDIUM = "medium" + HIGH = "high" + XHIGH = "xhigh" + MAX = "max" + + +class ConfigPlacement(StrEnum): + IN_REPO = "in_repo" + SIDECAR = "sidecar" + + +_EFFORT_ALIASES = { + "light": ThinkingEffort.LOW, + "extra_high": ThinkingEffort.XHIGH, + "extra high": ThinkingEffort.XHIGH, + "ultra": ThinkingEffort.MAX, +} +_EFFORT_ORDER = ( + ThinkingEffort.LOW, + ThinkingEffort.MEDIUM, + ThinkingEffort.HIGH, + ThinkingEffort.XHIGH, + ThinkingEffort.MAX, +) + + +def normalize_thinking_effort(value: str | ThinkingEffort) -> ThinkingEffort: + if isinstance(value, ThinkingEffort): + return value + normalized = value.strip().lower() + if normalized in _EFFORT_ALIASES: + return _EFFORT_ALIASES[normalized] + try: + return ThinkingEffort(normalized) + except ValueError as exc: + raise WorkspaceConfigError( + f"unsupported thinking effort {value!r}" + ) from exc + + +def canonical_json(value: Any) -> str: + """Return the only JSON encoding used for semantic digests.""" + + return json.dumps( + value, + ensure_ascii=False, + allow_nan=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def canonical_digest(value: Any) -> str: + return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest() + + +_SECRET_FIELD_NAMES = { + "access_token", + "api_key", + "authorization", + "client_secret", + "cookie", + "credential", + "credentials", + "password", + "private_key", + "refresh_token", + "secret", + "secret_value", + "token", +} +_SLOT_REFERENCE_KEYS = {"slot", "slot_id", "connection_slot", "secret_slot"} +_CLOUD_FORBIDDEN_FIELD_NAMES = { + "absolute_path", + "device_credential", + "device_identity", + "local_binding_id", + "local_connection_id", + "worktree_root", +} +_WINDOWS_ABSOLUTE_PATH = re.compile(r"^[a-zA-Z]:[\\/]") + + +def _normalized_field_name(value: str) -> str: + value = re.sub(r"(? bool: + if isinstance(value, str): + return value.startswith(("slot://", "connection://", "binding://")) + if not isinstance(value, dict) or not value: + return False + return set(map(_normalized_field_name, value)).issubset( + _SLOT_REFERENCE_KEYS + ) + + +def assert_manifest_secret_free(value: Any, path: str = "$") -> None: + """Reject secret values while allowing opaque slot references.""" + + if isinstance(value, dict): + for key, child in value.items(): + field_path = f"{path}.{key}" + normalized = _normalized_field_name(str(key)) + if normalized in _SECRET_FIELD_NAMES and not _is_slot_reference( + child + ): + raise SecretValueInManifestError( + f"secret-bearing field is forbidden in Bundle manifest: " + f"{field_path}" + ) + assert_manifest_secret_free(child, field_path) + return + if isinstance(value, (list, tuple)): + for index, child in enumerate(value): + assert_manifest_secret_free(child, f"{path}[{index}]") + + +def assert_cloud_projection_safe(value: Any, path: str = "$") -> None: + """Enforce the server-side projection schema's local-data denylist.""" + + if isinstance(value, dict): + for key, child in value.items(): + field_path = f"{path}.{key}" + normalized = _normalized_field_name(str(key)) + if normalized in _CLOUD_FORBIDDEN_FIELD_NAMES: + raise UnsafeCloudProjectionError( + f"device-local field is forbidden in Cloud projection: " + f"{field_path}" + ) + if ( + isinstance(child, str) + and ( + normalized == "path" + or normalized.endswith("_path") + or normalized.endswith("_root") + ) + and ( + child.startswith(("/", "~/", "\\\\")) + or _WINDOWS_ABSOLUTE_PATH.match(child) + ) + ): + raise UnsafeCloudProjectionError( + f"absolute path is forbidden in Cloud projection: " + f"{field_path}" + ) + assert_cloud_projection_safe(child, field_path) + return + if isinstance(value, (list, tuple)): + for index, child in enumerate(value): + assert_cloud_projection_safe(child, f"{path}[{index}]") + + +class _StrictFrozenModel(BaseModel): + model_config = ConfigDict( + extra="forbid", + frozen=True, + populate_by_name=True, + ) + + +class BundleMetadata(_StrictFrozenModel): + id: str = Field(min_length=1) + name: str = Field(min_length=1) + revision: int = Field(ge=1) + + +class ContextSource(_StrictFrozenModel): + id: str = Field(min_length=1) + kind: Literal[ + "bundle_asset", + "inline", + "connection_query", + "local_path_slot", + "artifact_ref", + "memory_scope", + ] + slot: str | None = None + path: str | None = None + content: str | None = None + query: dict[str, Any] | None = None + sharing: Literal["bundled", "reference_only", "authorized_artifact"] = ( + "reference_only" + ) + + @model_validator(mode="after") + def validate_source_shape(self) -> ContextSource: + if self.kind == "local_path_slot" and not self.slot: + raise ValueError("local_path_slot requires slot") + if self.kind == "local_path_slot" and self.path is not None: + raise ValueError("local_path_slot cannot contain a physical path") + if self.kind == "bundle_asset": + if not self.path or not self.path.startswith("bundle://"): + raise ValueError( + "bundle_asset path must use a bundle:// logical URI" + ) + if self.kind == "inline" and self.content is None: + raise ValueError("inline context requires content") + if self.kind == "connection_query" and self.query is None: + raise ValueError("connection_query requires query") + return self + + +class SkillAssignment(_StrictFrozenModel): + ref: str = Field(min_length=1) + assign_to: tuple[str, ...] = Field( + default_factory=tuple, + alias="assignTo", + ) + + @field_validator("ref") + @classmethod + def validate_logical_ref(cls, value: str) -> str: + if not value.startswith(("bundle://", "registry://")): + raise ValueError("skill ref must use bundle:// or registry://") + return value + + +class ConnectorRequirement(_StrictFrozenModel): + id: str = Field(min_length=1) + connector: str = Field(min_length=1) + connection_slot: str = Field(min_length=1, alias="connectionSlot") + required_grants: tuple[str, ...] = Field( + default_factory=tuple, + alias="requiredGrants", + ) + + +class McpServerRequirement(_StrictFrozenModel): + id: str = Field(min_length=1) + definition: str = Field(min_length=1) + secret_slots: tuple[str, ...] = Field( + default_factory=tuple, + alias="secretSlots", + ) + assign_to: tuple[str, ...] = Field( + default_factory=tuple, + alias="assignTo", + ) + + @field_validator("definition") + @classmethod + def validate_logical_definition(cls, value: str) -> str: + if not value.startswith(("bundle://", "registry://")): + raise ValueError( + "MCP definition must use bundle:// or registry://" + ) + return value + + +class AgentProfile(_StrictFrozenModel): + id: str = Field(min_length=1) + role: str = Field(min_length=1) + model_profile: str = Field(min_length=1, alias="modelProfile") + + +class ModelProfile(_StrictFrozenModel): + model_ref: str = Field(min_length=1, alias="modelRef") + thinking_effort: ThinkingEffort = Field( + default=ThinkingEffort.MEDIUM, + alias="thinkingEffort", + ) + + @field_validator("thinking_effort", mode="before") + @classmethod + def normalize_effort(cls, value: Any) -> ThinkingEffort: + if not isinstance(value, (str, ThinkingEffort)): + raise ValueError("thinkingEffort must be a string") + return normalize_thinking_effort(value) + + @field_validator("model_ref") + @classmethod + def validate_model_ref(cls, value: str) -> str: + if not value.startswith("provider://"): + raise ValueError("modelRef must use provider://") + return value + + +class PermissionRule(_StrictFrozenModel): + action: str = Field(min_length=1) + effect: Literal["allow", "prompt", "deny"] + + +class PermissionProfile(_StrictFrozenModel): + profile: Literal[ + "request_approval", + "auto_review", + "workspace_write", + "full_access", + ] = "request_approval" + rules: tuple[PermissionRule, ...] = Field(default_factory=tuple) + + +class GitPolicy(_StrictFrozenModel): + enabled: bool = True + checkpoint_policy: str = Field( + default="user_and_run_terminal", + alias="checkpointPolicy", + ) + agent_isolation: Literal["worktree"] = Field( + default="worktree", + alias="agentIsolation", + ) + remote_policy: Literal["deny", "prompt", "allow"] = Field( + default="prompt", + alias="remotePolicy", + ) + + +class BundleSpec(_StrictFrozenModel): + instructions: dict[str, str] = Field(default_factory=dict) + context: tuple[ContextSource, ...] = Field(default_factory=tuple) + skills: tuple[SkillAssignment, ...] = Field(default_factory=tuple) + connectors: tuple[ConnectorRequirement, ...] = Field(default_factory=tuple) + mcp_servers: tuple[McpServerRequirement, ...] = Field( + default_factory=tuple, + alias="mcpServers", + ) + agents: tuple[AgentProfile, ...] = Field(default_factory=tuple) + models: dict[str, ModelProfile] = Field(default_factory=dict) + permissions: PermissionProfile = Field(default_factory=PermissionProfile) + git: GitPolicy = Field(default_factory=GitPolicy) + + @field_validator("instructions") + @classmethod + def validate_instruction_refs( + cls, value: dict[str, str] + ) -> dict[str, str]: + invalid = [ + role + for role, ref in value.items() + if not ref.startswith("bundle://") + ] + if invalid: + raise ValueError( + "instruction refs must use bundle://: " + + ", ".join(sorted(invalid)) + ) + return value + + @model_validator(mode="after") + def validate_references(self) -> BundleSpec: + agent_ids = {agent.id for agent in self.agents} + if len(agent_ids) != len(self.agents): + raise ValueError("agent ids must be unique") + context_ids = {source.id for source in self.context} + if len(context_ids) != len(self.context): + raise ValueError("context source ids must be unique") + connector_ids = {item.id for item in self.connectors} + if len(connector_ids) != len(self.connectors): + raise ValueError("connector ids must be unique") + mcp_ids = {item.id for item in self.mcp_servers} + if len(mcp_ids) != len(self.mcp_servers): + raise ValueError("MCP server ids must be unique") + missing_profiles = { + agent.model_profile + for agent in self.agents + if agent.model_profile not in self.models + } + if missing_profiles: + raise ValueError( + "agents reference missing model profiles: " + + ", ".join(sorted(missing_profiles)) + ) + assigned_agents = { + agent_id + for assignment in (*self.skills, *self.mcp_servers) + for agent_id in assignment.assign_to + } + unknown_agents = assigned_agents - agent_ids + if unknown_agents: + raise ValueError( + "assignTo references unknown agents: " + + ", ".join(sorted(unknown_agents)) + ) + return self + + +class WorkforceBundleManifest(_StrictFrozenModel): + api_version: Literal["eigent.ai/v1alpha1"] = Field(alias="apiVersion") + kind: Literal["WorkforceBundle"] + metadata: BundleMetadata + spec: BundleSpec + + @model_validator(mode="after") + def reject_secret_values(self) -> WorkforceBundleManifest: + assert_manifest_secret_free(self.canonical_payload()) + return self + + def canonical_payload(self) -> dict[str, Any]: + return self.model_dump(by_alias=True, exclude_none=True, mode="json") + + @property + def digest(self) -> str: + return canonical_digest(self.canonical_payload()) + + @property + def revision_id(self) -> str: + return f"{self.metadata.id}@{self.metadata.revision}" + + +@dataclass(frozen=True) +class EffortResolution: + requested: ThinkingEffort + effective: ThinkingEffort + provider_value: str + capability_revision: str + remapped: bool + + +@dataclass(frozen=True) +class ProviderModelCapability: + supported_efforts: tuple[ThinkingEffort, ...] + default_effort: ThinkingEffort + provider_mapping: dict[ThinkingEffort, str] + capability_revision: str + dynamic_model: bool = False + + def __post_init__(self) -> None: + supported = tuple(dict.fromkeys(self.supported_efforts)) + if not supported: + raise WorkspaceConfigError( + "provider capability must support at least one effort" + ) + if self.default_effort not in supported: + raise WorkspaceConfigError( + "provider default effort must be supported" + ) + missing = set(supported) - set(self.provider_mapping) + if missing: + raise WorkspaceConfigError( + "provider mapping is missing efforts: " + + ", ".join(sorted(item.value for item in missing)) + ) + if not self.capability_revision.strip(): + raise WorkspaceConfigError( + "provider capability revision is required" + ) + object.__setattr__(self, "supported_efforts", supported) + object.__setattr__( + self, + "provider_mapping", + MappingProxyType(dict(self.provider_mapping)), + ) + + def resolve( + self, + requested: str | ThinkingEffort | None, + *, + allow_dynamic_remap: bool = False, + ) -> EffortResolution: + normalized = ( + normalize_thinking_effort(requested) + if requested is not None + else self.default_effort + ) + effective = normalized + if effective not in self.supported_efforts: + if not (self.dynamic_model and allow_dynamic_remap): + raise UnsupportedThinkingEffortError( + f"effort {effective.value!r} is not supported by " + f"capability {self.capability_revision!r}" + ) + requested_index = _EFFORT_ORDER.index(effective) + effective = min( + self.supported_efforts, + key=lambda candidate: ( + abs(_EFFORT_ORDER.index(candidate) - requested_index), + _EFFORT_ORDER.index(candidate) > requested_index, + ), + ) + return EffortResolution( + requested=normalized, + effective=effective, + provider_value=self.provider_mapping[effective], + capability_revision=self.capability_revision, + remapped=effective is not normalized, + ) + + +class ResolvedContextSource(_StrictFrozenModel): + id: str + kind: str + logical_uri: str | None = None + slot_id: str | None = None + absolute_path: str | None = None + root_fingerprint_digest: str | None = None + + def cloud_projection(self) -> dict[str, Any]: + return self.model_dump( + include={ + "id", + "kind", + "logical_uri", + "slot_id", + "root_fingerprint_digest", + }, + exclude_none=True, + mode="json", + ) + + +class ResolvedConnectorBinding(_StrictFrozenModel): + connector_id: str + slot_id: str + local_binding_id: str | None = None + required_grants: tuple[str, ...] = Field(default_factory=tuple) + + def cloud_projection(self) -> dict[str, Any]: + return self.model_dump( + include={"connector_id", "slot_id", "required_grants"}, + mode="json", + ) + + +class WorktreeMaterialization(_StrictFrozenModel): + repository_id: str + logical_worktree_role: str + absolute_path: str | None = None + base_commit: str | None = None + + def cloud_projection(self) -> dict[str, Any]: + return self.model_dump( + include={ + "repository_id", + "logical_worktree_role", + "base_commit", + }, + exclude_none=True, + mode="json", + ) + + +class LocalMaterialization(_StrictFrozenModel): + context_sources: tuple[ResolvedContextSource, ...] = Field( + default_factory=tuple + ) + connector_bindings: tuple[ResolvedConnectorBinding, ...] = Field( + default_factory=tuple + ) + worktree: WorktreeMaterialization | None = None + + def cloud_projection(self) -> dict[str, Any]: + return { + "context_sources": [ + source.cloud_projection() for source in self.context_sources + ], + "connector_bindings": [ + binding.cloud_projection() + for binding in self.connector_bindings + ], + "worktree": ( + self.worktree.cloud_projection() if self.worktree else None + ), + } + + +class EffectiveEnvironmentSpec(_StrictFrozenModel): + spec_id: str + owner_type: Literal["run", "run_attempt"] + owner_id: str + bundle_revision_id: str + manifest_digest: str + semantic_spec: dict[str, Any] + semantic_spec_digest: str + local_materialization: LocalMaterialization + local_materialization_digest: str + permission_profile_revision: str + thinking_effort_requested: ThinkingEffort + thinking_effort_effective: ThinkingEffort + provider_value: str + provider_capability_revision: str + redaction_schema_version: int = 1 + + @classmethod + def create( + cls, + *, + owner_type: Literal["run", "run_attempt"], + owner_id: str, + manifest: WorkforceBundleManifest, + semantic_spec: dict[str, Any], + local_materialization: LocalMaterialization, + permission_profile_revision: str, + effort: EffortResolution, + ) -> EffectiveEnvironmentSpec: + assert_manifest_secret_free(semantic_spec) + semantic_digest = canonical_digest(semantic_spec) + local_payload = local_materialization.model_dump( + exclude_none=True, + mode="json", + ) + local_digest = canonical_digest(local_payload) + identity = { + "owner_type": owner_type, + "owner_id": owner_id, + "bundle_revision_id": manifest.revision_id, + "semantic_spec_digest": semantic_digest, + "local_materialization_digest": local_digest, + "permission_profile_revision": permission_profile_revision, + "thinking_effort_requested": effort.requested.value, + "thinking_effort_effective": effort.effective.value, + "provider_capability_revision": effort.capability_revision, + } + spec_id = f"envspec_{canonical_digest(identity)}" + return cls( + spec_id=spec_id, + owner_type=owner_type, + owner_id=owner_id, + bundle_revision_id=manifest.revision_id, + manifest_digest=manifest.digest, + semantic_spec=semantic_spec, + semantic_spec_digest=semantic_digest, + local_materialization=local_materialization, + local_materialization_digest=local_digest, + permission_profile_revision=permission_profile_revision, + thinking_effort_requested=effort.requested, + thinking_effort_effective=effort.effective, + provider_value=effort.provider_value, + provider_capability_revision=effort.capability_revision, + ) + + def local_payload(self) -> dict[str, Any]: + return self.model_dump(exclude_none=True, mode="json") + + @property + def digest(self) -> str: + return canonical_digest(self.local_payload()) + + def cloud_projection(self) -> dict[str, Any]: + payload = { + "schema_version": 1, + "owner_type": self.owner_type, + "owner_id": self.owner_id, + "bundle_revision_id": self.bundle_revision_id, + "manifest_digest": self.manifest_digest, + "semantic_spec": self.semantic_spec, + "semantic_spec_digest": self.semantic_spec_digest, + "local_projection": self.local_materialization.cloud_projection(), + "permission_profile_revision": self.permission_profile_revision, + "thinking_effort_requested": self.thinking_effort_requested.value, + "thinking_effort_effective": self.thinking_effort_effective.value, + "provider_capability_revision": ( + self.provider_capability_revision + ), + "redaction_schema_version": self.redaction_schema_version, + } + assert_cloud_projection_safe(payload) + projection = { + **payload, + "projection_digest": canonical_digest(payload), + } + assert_cloud_projection_safe(projection) + return projection diff --git a/backend/app/workspace_config/resolver.py b/backend/app/workspace_config/resolver.py new file mode 100644 index 000000000..f201a56e3 --- /dev/null +++ b/backend/app/workspace_config/resolver.py @@ -0,0 +1,81 @@ +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +"""Deterministic Phase-1 EnvironmentSpec resolver.""" + +from __future__ import annotations + +from typing import Any, Literal + +from app.workspace_config.models import ( + EffectiveEnvironmentSpec, + LocalMaterialization, + ProviderModelCapability, + ThinkingEffort, + WorkforceBundleManifest, + canonical_digest, +) + + +class EnvironmentConfigResolver: + """Resolve immutable semantic/local snapshots without reading secrets.""" + + def resolve( + self, + *, + manifest: WorkforceBundleManifest, + owner_type: Literal["run", "run_attempt"], + owner_id: str, + local_materialization: LocalMaterialization, + provider_capability: ProviderModelCapability, + model_profile: str = "default", + thinking_effort_override: str | ThinkingEffort | None = None, + allow_dynamic_effort_remap: bool = False, + runtime_capability_manifest: dict[str, Any] | None = None, + ) -> EffectiveEnvironmentSpec: + try: + profile = manifest.spec.models[model_profile] + except KeyError as exc: + raise ValueError( + f"Bundle does not define model profile {model_profile!r}" + ) from exc + requested_effort = ( + thinking_effort_override + if thinking_effort_override is not None + else profile.thinking_effort + ) + effort = provider_capability.resolve( + requested_effort, + allow_dynamic_remap=allow_dynamic_effort_remap, + ) + permission_payload = manifest.spec.permissions.model_dump( + by_alias=True, + mode="json", + ) + permission_revision = canonical_digest(permission_payload) + semantic_spec = { + "bundle": manifest.canonical_payload(), + "selected_model_profile": model_profile, + "model_ref": profile.model_ref, + "runtime_capability_manifest": runtime_capability_manifest or {}, + } + return EffectiveEnvironmentSpec.create( + owner_type=owner_type, + owner_id=owner_id, + manifest=manifest, + semantic_spec=semantic_spec, + local_materialization=local_materialization, + permission_profile_revision=permission_revision, + effort=effort, + ) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 625debd24..fa42d430e 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -21,6 +21,7 @@ dependencies = [ "numpy>=1.23.0,<2.0.0", "debugpy>=1.8.17", "qdrant-client>=1.16.2", + "pyyaml>=6.0.3", "opentelemetry-api>=1.34.1", "opentelemetry-sdk>=1.34.1", "opentelemetry-exporter-otlp-proto-http>=1.34.1", diff --git a/backend/tests/app/run_journal/test_store.py b/backend/tests/app/run_journal/test_store.py index 23ae6d030..de6354e9a 100644 --- a/backend/tests/app/run_journal/test_store.py +++ b/backend/tests/app/run_journal/test_store.py @@ -21,6 +21,7 @@ import pytest from app.run_journal import ( SCHEMA_VERSION, + AttemptEnvironmentBinding, CloudRunEventReplica, CloudRunReplica, EventRecorder, @@ -32,6 +33,13 @@ from app.run_journal import ( RunNotFoundError, SQLiteRunJournal, ) +from app.workspace_config import ( + EnvironmentConfigResolver, + LocalMaterialization, + ProviderModelCapability, + ThinkingEffort, + parse_workforce_manifest, +) @pytest.fixture @@ -63,9 +71,190 @@ def test_initializes_schema_and_durability_pragmas(journal): "run_event_sync_outbox", "tool_calls", "approvals", + "workspace_config_revisions", + "workspace_config_materializations", + "effective_environment_specs", } <= tables +def _persist_environment_spec(journal, *, owner_id: str = "run-1"): + manifest = parse_workforce_manifest( + """ +apiVersion: eigent.ai/v1alpha1 +kind: WorkforceBundle +metadata: + id: bundle_test + name: Test Bundle + revision: 1 +spec: + agents: + - id: coordinator + role: coordinator + modelProfile: default + models: + default: + modelRef: provider://default + thinkingEffort: medium +""" + ) + revision = journal.put_workspace_config_revision( + revision_id=manifest.revision_id, + space_id="space-1", + bundle_id=manifest.metadata.id, + revision_number=manifest.metadata.revision, + config_placement="sidecar", + manifest=manifest.canonical_payload(), + created_by="user-1", + now=1, + ) + capability = ProviderModelCapability( + supported_efforts=tuple(ThinkingEffort), + default_effort=ThinkingEffort.MEDIUM, + provider_mapping={effort: effort.value for effort in ThinkingEffort}, + capability_revision="capability-v1", + ) + spec = EnvironmentConfigResolver().resolve( + manifest=manifest, + owner_type="run", + owner_id=owner_id, + local_materialization=LocalMaterialization(), + provider_capability=capability, + ) + persisted = journal.put_effective_environment_spec(spec, now=2) + return revision, spec, persisted + + +def test_workspace_config_and_environment_spec_are_immutable(journal): + revision, spec, persisted = _persist_environment_spec(journal) + + assert revision.manifest_digest == spec.manifest_digest + assert persisted.environment_spec_id == spec.spec_id + assert persisted.environment_spec_digest == spec.digest + assert ( + persisted.redacted_spec["projection_digest"] + == persisted.projection_digest + ) + assert journal.put_effective_environment_spec(spec, now=3) == persisted + + with pytest.raises(IdempotencyConflictError, match="config revision"): + journal.put_workspace_config_revision( + revision_id=revision.revision_id, + space_id=revision.space_id, + bundle_id=revision.bundle_id, + revision_number=revision.revision_number, + config_placement=revision.config_placement, + manifest={**revision.manifest, "unexpected": True}, + created_by=revision.created_by, + ) + + +def test_workspace_config_revision_lifecycle_uses_version_cas(journal): + manifest = parse_workforce_manifest( + """ +apiVersion: eigent.ai/v1alpha1 +kind: WorkforceBundle +metadata: + id: bundle_lifecycle + name: Lifecycle Bundle + revision: 1 +spec: + models: + default: + modelRef: provider://default + thinkingEffort: medium +""" + ) + draft = journal.put_workspace_config_revision( + revision_id=manifest.revision_id, + space_id="space-1", + bundle_id=manifest.metadata.id, + revision_number=manifest.metadata.revision, + config_placement="sidecar", + manifest=manifest.canonical_payload(), + status="draft", + created_by="user-1", + ) + + validated = journal.transition_workspace_config_revision( + draft.revision_id, + expected_version=0, + status="validated", + ) + published = journal.transition_workspace_config_revision( + draft.revision_id, + expected_version=1, + status="published", + ) + + assert (draft.status, draft.version) == ("draft", 0) + assert (validated.status, validated.version) == ("validated", 1) + assert (published.status, published.version) == ("published", 2) + assert published.manifest == draft.manifest + with pytest.raises(OptimisticConcurrencyError): + journal.transition_workspace_config_revision( + draft.revision_id, + expected_version=1, + status="deprecated", + ) + with pytest.raises(InvalidRunTransitionError): + journal.transition_workspace_config_revision( + draft.revision_id, + expected_version=2, + status="validated", + ) + + +def test_attempt_binds_environment_once_and_emits_resolved_values(journal): + _, spec, _ = _persist_environment_spec(journal) + journal.ensure_run( + run_id="run-1", + project_id="project-1", + status="pending", + ) + environment = AttemptEnvironmentBinding( + environment_spec_id=spec.spec_id, + environment_spec_digest=spec.digest, + bundle_revision_id=spec.bundle_revision_id, + permission_profile_revision=spec.permission_profile_revision, + thinking_effort_requested=spec.thinking_effort_requested.value, + thinking_effort_effective=spec.thinking_effort_effective.value, + provider_capability_revision=spec.provider_capability_revision, + ) + + attempt = journal.create_run_attempt( + "run-1", + request_id="request-1", + reason="initial_execution", + environment=environment, + attempt_id="attempt-1", + ) + replay = journal.create_run_attempt( + "run-1", + request_id="request-1", + reason="initial_execution", + environment=environment, + attempt_id="attempt-1", + ) + + assert replay == attempt + assert attempt.environment_spec_id == spec.spec_id + assert attempt.environment_spec_digest == spec.digest + assert attempt.thinking_effort_requested == "medium" + assert attempt.thinking_effort_effective == "medium" + event = journal.list_events("run-1")[0] + assert event.event_type == "run.attempt_created" + assert event.payload["environment_spec_id"] == spec.spec_id + assert event.payload["thinking_effort_effective"] == "medium" + + with pytest.raises(IdempotencyConflictError, match="environment"): + journal.create_run_attempt( + "run-1", + request_id="request-1", + reason="initial_execution", + environment=None, + ) + + def test_ensure_run_rejects_policy_and_deadline_drift(journal): first = journal.ensure_run( run_id="run-1", diff --git a/backend/tests/app/workspace_config/__init__.py b/backend/tests/app/workspace_config/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/backend/tests/app/workspace_config/__init__.py @@ -0,0 +1 @@ + diff --git a/backend/tests/app/workspace_config/test_models.py b/backend/tests/app/workspace_config/test_models.py new file mode 100644 index 000000000..b306c31e9 --- /dev/null +++ b/backend/tests/app/workspace_config/test_models.py @@ -0,0 +1,246 @@ +from __future__ import annotations + +import json + +import pytest +from pydantic import ValidationError + +from app.workspace_config import ( + EnvironmentConfigResolver, + LocalMaterialization, + ProviderModelCapability, + ResolvedConnectorBinding, + ResolvedContextSource, + SecretValueInManifestError, + ThinkingEffort, + UnsafeCloudProjectionError, + UnsupportedThinkingEffortError, + WorktreeMaterialization, + canonical_digest, + parse_workforce_manifest, +) + +MANIFEST_YAML = """ +apiVersion: eigent.ai/v1alpha1 +kind: WorkforceBundle +metadata: + id: bundle_product_research + name: Product Research Workforce + revision: 7 +spec: + instructions: + coordinator: bundle://instructions/coordinator.md + context: + - id: product_docs + kind: local_path_slot + slot: product_docs_folder + sharing: reference_only + - id: research_policy + kind: bundle_asset + path: bundle://context/README.md + sharing: bundled + skills: + - ref: registry://skills/web-research@2.1.0 + assignTo: [coordinator] + connectors: + - id: source_repository + connector: github + connectionSlot: github_readonly + requiredGrants: [repository.read] + mcpServers: + - id: issue_tracker + definition: registry://mcp/linear@1.4.0 + secretSlots: [LINEAR_API_TOKEN] + assignTo: [coordinator] + agents: + - id: coordinator + role: coordinator + modelProfile: default + models: + default: + modelRef: provider://default + thinkingEffort: medium + permissions: + profile: request_approval + rules: + - action: connector.read + effect: allow + git: + enabled: true + checkpointPolicy: user_and_run_terminal + agentIsolation: worktree + remotePolicy: prompt +""" + + +def _capability() -> ProviderModelCapability: + return ProviderModelCapability( + supported_efforts=( + ThinkingEffort.LOW, + ThinkingEffort.MEDIUM, + ThinkingEffort.HIGH, + ), + default_effort=ThinkingEffort.MEDIUM, + provider_mapping={ + ThinkingEffort.LOW: "low", + ThinkingEffort.MEDIUM: "medium", + ThinkingEffort.HIGH: "high", + }, + capability_revision="provider-capability-v3", + ) + + +def test_manifest_is_strict_canonical_and_digest_stable(): + first = parse_workforce_manifest(MANIFEST_YAML) + reordered = parse_workforce_manifest( + MANIFEST_YAML.replace( + " name: Product Research Workforce\n revision: 7", + " revision: 7\n name: Product Research Workforce", + ) + ) + + assert first == reordered + assert first.digest == reordered.digest + assert first.revision_id == "bundle_product_research@7" + assert ( + first.spec.models["default"].thinking_effort is ThinkingEffort.MEDIUM + ) + + +def test_manifest_rejects_secret_value_even_inside_freeform_instructions(): + with pytest.raises(SecretValueInManifestError, match="api_key"): + parse_workforce_manifest( + MANIFEST_YAML.replace( + "coordinator: bundle://instructions/coordinator.md", + "api_key: sk-must-not-enter-a-bundle", + ) + ) + + +def test_manifest_rejects_physical_path_for_local_slot(): + with pytest.raises(ValidationError, match="physical path"): + parse_workforce_manifest( + MANIFEST_YAML.replace( + " sharing: reference_only", + " path: /Users/alice/private\n" + " sharing: reference_only", + 1, + ) + ) + + +def test_thinking_effort_aliases_are_normalized_and_unsupported_is_explicit(): + manifest = parse_workforce_manifest( + MANIFEST_YAML.replace( + "thinkingEffort: medium", "thinkingEffort: light" + ) + ) + assert ( + manifest.spec.models["default"].thinking_effort is ThinkingEffort.LOW + ) + + with pytest.raises(UnsupportedThinkingEffortError, match="xhigh"): + _capability().resolve(ThinkingEffort.XHIGH) + + +def test_dynamic_provider_remap_is_opt_in_and_reported(): + capability = ProviderModelCapability( + supported_efforts=(ThinkingEffort.LOW, ThinkingEffort.HIGH), + default_effort=ThinkingEffort.LOW, + provider_mapping={ + ThinkingEffort.LOW: "minimal", + ThinkingEffort.HIGH: "deep", + }, + capability_revision="dynamic-v1", + dynamic_model=True, + ) + + with pytest.raises(UnsupportedThinkingEffortError): + capability.resolve(ThinkingEffort.MEDIUM) + resolved = capability.resolve( + ThinkingEffort.MEDIUM, + allow_dynamic_remap=True, + ) + + assert resolved.requested is ThinkingEffort.MEDIUM + assert resolved.effective is ThinkingEffort.LOW + assert resolved.provider_value == "minimal" + assert resolved.remapped is True + + +def test_cloud_projection_redacts_local_paths_and_binding_ids(): + manifest = parse_workforce_manifest(MANIFEST_YAML) + local = LocalMaterialization( + context_sources=( + ResolvedContextSource( + id="product_docs", + kind="local_path_slot", + slot_id="product_docs_folder", + absolute_path="/Users/alice/company/private", + root_fingerprint_digest="root-digest", + ), + ), + connector_bindings=( + ResolvedConnectorBinding( + connector_id="github", + slot_id="github_readonly", + local_binding_id="connection-secret-device-id", + required_grants=("repository.read",), + ), + ), + worktree=WorktreeMaterialization( + repository_id="repo-1", + logical_worktree_role="run_integration", + absolute_path="/Users/alice/.eigent/worktrees/run-1", + base_commit="abc123", + ), + ) + spec = EnvironmentConfigResolver().resolve( + manifest=manifest, + owner_type="run", + owner_id="run-1", + local_materialization=local, + provider_capability=_capability(), + ) + + local_json = json.dumps(spec.local_payload()) + cloud = spec.cloud_projection() + cloud_json = json.dumps(cloud) + assert "/Users/alice/company/private" in local_json + assert "connection-secret-device-id" in local_json + assert "/Users/alice/company/private" not in cloud_json + assert "connection-secret-device-id" not in cloud_json + assert "/Users/alice/.eigent/worktrees/run-1" not in cloud_json + assert cloud["local_projection"]["context_sources"] == [ + { + "id": "product_docs", + "kind": "local_path_slot", + "slot_id": "product_docs_folder", + "root_fingerprint_digest": "root-digest", + } + ] + projection_body = { + key: value + for key, value in cloud.items() + if key != "projection_digest" + } + assert cloud["projection_digest"] == canonical_digest(projection_body) + + +def test_cloud_projection_rejects_local_fields_in_semantic_capabilities(): + manifest = parse_workforce_manifest(MANIFEST_YAML) + spec = EnvironmentConfigResolver().resolve( + manifest=manifest, + owner_type="run", + owner_id="run-1", + local_materialization=LocalMaterialization(), + provider_capability=_capability(), + runtime_capability_manifest={ + "browser": { + "socket_path": "/Users/alice/.eigent/browser.sock", + } + }, + ) + + with pytest.raises(UnsafeCloudProjectionError, match="absolute path"): + spec.cloud_projection() diff --git a/backend/uv.lock b/backend/uv.lock index c26df7445..55478918b 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -229,6 +229,7 @@ dependencies = [ { name = "pydantic-i18n" }, { name = "pydash" }, { name = "python-dotenv" }, + { name = "pyyaml" }, { name = "qdrant-client" }, { name = "truststore" }, { name = "uvicorn", extra = ["standard"] }, @@ -261,6 +262,7 @@ requires-dist = [ { name = "pydantic-i18n", specifier = ">=0.4.5" }, { name = "pydash", specifier = ">=8.0.5" }, { name = "python-dotenv", specifier = ">=1.1.0" }, + { name = "pyyaml", specifier = ">=6.0.3" }, { name = "qdrant-client", specifier = ">=1.16.2" }, { name = "truststore", specifier = ">=0.10.0" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.34.2" }, @@ -2686,9 +2688,9 @@ wheels = [ name = "truststore" version = "0.10.4" source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.730Z" } +sdist = { url = "https://files.pythonhosted.org/packages/53/a3/1585216310e344e8102c22482f6060c7a6ea0322b63e026372e6dcefcfd6/truststore-0.10.4.tar.gz", hash = "sha256:9d91bd436463ad5e4ee4aba766628dd6cd7010cf3e2461756b3303710eebc301", size = 26169, upload-time = "2025-08-12T18:49:02.73Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.460Z" }, + { url = "https://files.pythonhosted.org/packages/19/97/56608b2249fe206a67cd573bc93cd9896e1efb9e98bce9c163bcdc704b88/truststore-0.10.4-py3-none-any.whl", hash = "sha256:adaeaecf1cbb5f4de3b1959b42d41f6fab57b2b1666adb59e89cb0b53361d981", size = 18660, upload-time = "2025-08-12T18:49:01.46Z" }, ] [[package]]