diff --git a/backend/app/controller/workspace_bundle_controller.py b/backend/app/controller/workspace_bundle_controller.py new file mode 100644 index 000000000..e812ea1ef --- /dev/null +++ b/backend/app/controller/workspace_bundle_controller.py @@ -0,0 +1,336 @@ +"""Capability-protected local API for review-first Bundle installation.""" + +from __future__ import annotations + +import os +from dataclasses import asdict +from pathlib import Path +from typing import Annotated, Literal + +from fastapi import APIRouter, Depends, Header, HTTPException, Request +from pydantic import BaseModel, Field + +from app.auth import require_local_control_principal +from app.component.environment import env +from app.run_journal import ( + IdempotencyConflictError, + InvalidRunTransitionError, + OptimisticConcurrencyError, + configured_run_journal_path, + get_default_run_journal, +) +from app.router_layer.hands_resolver import get_environment_hands +from app.utils.workspace_resolver import get_workspace_resolver +from app.workspace_bundle import ( + HttpWorkspaceBundleCloudTransport, + WorkspaceBundleBindingsIncomplete, + WorkspaceBundleCloudError, + WorkspaceBundleInstallError, + WorkspaceBundleInstaller, +) +from app.workspace_config import ConfigPlacement +from app.workspace_git import ConfigurationRepositoryService + +router = APIRouter(dependencies=[Depends(require_local_control_principal)]) + + +class BundleProposalBody(BaseModel): + proposal_id: str = Field(min_length=1, max_length=128) + request_id: str = Field(min_length=1, max_length=128) + space_id: str = Field(min_length=1, max_length=256) + bundle_id: str = Field(min_length=1, max_length=128) + revision_id: str = Field(min_length=1, max_length=128) + config_placement: Literal["in_repo", "sidecar"] = "sidecar" + + +class BundleDecisionBody(BaseModel): + expected_version: int = Field(ge=0) + approved: bool + actor_id: str = Field(min_length=1, max_length=200) + + +class BundleConnectorBindingBody(BaseModel): + expected_version: int = Field(ge=0) + slot_id: str = Field(min_length=1, max_length=255) + connector_id: str = Field(min_length=1, max_length=255) + connection_id: str = Field(min_length=1, max_length=255) + actor_id: str = Field(min_length=1, max_length=200) + + +class BundleLocalPathBindingBody(BaseModel): + expected_version: int = Field(ge=0) + slot_id: str = Field(min_length=1, max_length=255) + local_path: str = Field(min_length=1, max_length=4096) + actor_id: str = Field(min_length=1, max_length=200) + + +class BundleScriptApprovalBody(BaseModel): + expected_version: int = Field(ge=0) + action_id: str = Field(min_length=1, max_length=1024) + actor_id: str = Field(min_length=1, max_length=200) + + +class BundleMaterializeBody(BaseModel): + expected_version: int = Field(ge=0) + email: str = Field(min_length=1, max_length=512) + user_id: str | int | None = None + actor_id: str = Field(min_length=1, max_length=200) + allow_content_repository_init: bool = False + + +def _configuration_repository() -> ConfigurationRepositoryService: + journal = get_default_run_journal() + return ConfigurationRepositoryService( + journal, + state_root=configured_run_journal_path().parent / "workspace-git", + ) + + +def _installer(cloud=None) -> WorkspaceBundleInstaller: + return WorkspaceBundleInstaller( + get_default_run_journal(), + _configuration_repository(), + cloud, + ) + + +def _cloud(authorization: str) -> HttpWorkspaceBundleCloudTransport: + # The bearer credential may only be sent to the process-owned SERVER_URL. + # Renderer input cannot choose or override its destination. + server_url = env("SERVER_URL", "").strip() + if not server_url: + raise WorkspaceBundleInstallError("SERVER_URL is not configured") + return HttpWorkspaceBundleCloudTransport( + server_url=server_url, + authorization=authorization, + desktop_instance_id=os.environ.get( + "EIGENT_DESKTOP_INSTANCE_ID", "" + ), + ) + + +def _payload(proposal_id: str) -> dict: + journal = get_default_run_journal() + proposal = journal.get_workspace_bundle_install_proposal(proposal_id) + if proposal is None: + raise HTTPException( + status_code=404, + detail={"code": "bundle_install_proposal_not_found"}, + ) + return { + "proposal": asdict(proposal), + "bindings": [ + asdict(item) + for item in journal.list_workspace_bundle_local_bindings( + proposal_id + ) + ], + } + + +def _error(exc: Exception) -> HTTPException: + if isinstance(exc, HTTPException): + return exc + if isinstance(exc, WorkspaceBundleCloudError): + return HTTPException( + status_code=exc.status_code, + detail={"code": "bundle_cloud_error", "upstream": exc.detail}, + ) + if isinstance(exc, WorkspaceBundleBindingsIncomplete): + return HTTPException( + status_code=409, + detail={ + "code": "bundle_bindings_incomplete", + "missing_slots": list(exc.missing_slots), + }, + ) + if isinstance( + exc, + ( + IdempotencyConflictError, + InvalidRunTransitionError, + OptimisticConcurrencyError, + ), + ): + return HTTPException( + status_code=409, + detail={"code": "bundle_install_conflict", "message": str(exc)}, + ) + if isinstance(exc, (WorkspaceBundleInstallError, ValueError)): + return HTTPException( + status_code=422, + detail={"code": "bundle_install_invalid", "message": str(exc)}, + ) + return HTTPException( + status_code=500, + detail={"code": "bundle_install_failed"}, + ) + + +def _authorized_local_path(request: Request, value: str) -> Path: + hands = getattr(request.state, "hands", None) or get_environment_hands() + validator = getattr(hands, "validate_workspace_binding_path", None) + if validator is not None: + ok, reason = validator(value) + if not ok: + raise WorkspaceBundleInstallError( + f"Local path is not allowed: {reason or 'path_not_allowed'}" + ) + else: + can_access = getattr(hands, "can_access_filesystem", None) + if can_access is None or not can_access(value): + raise WorkspaceBundleInstallError( + "Local path is outside the Desktop filesystem capability" + ) + return Path(value) + + +@router.post("/workspace-bundles/install-proposals") +async def propose_bundle_install( + body: BundleProposalBody, + authorization: Annotated[str, Header(alias="Authorization")], +) -> dict: + cloud = None + try: + cloud = _cloud(authorization) + await _installer(cloud).propose( + proposal_id=body.proposal_id, + request_id=body.request_id, + space_id=body.space_id, + bundle_id=body.bundle_id, + revision_id=body.revision_id, + config_placement=ConfigPlacement(body.config_placement), + ) + return _payload(body.proposal_id) + except Exception as exc: + raise _error(exc) from exc + finally: + if cloud is not None: + await cloud.close() + + +@router.get("/workspace-bundles/install-proposals/{proposal_id}") +async def get_bundle_install_proposal(proposal_id: str) -> dict: + return _payload(proposal_id) + + +@router.post("/workspace-bundles/install-proposals/{proposal_id}/decision") +async def decide_bundle_install( + proposal_id: str, body: BundleDecisionBody +) -> dict: + try: + _installer().decide( + proposal_id, + expected_version=body.expected_version, + approved=body.approved, + decided_by=body.actor_id, + ) + return _payload(proposal_id) + except Exception as exc: + raise _error(exc) from exc + + +@router.post( + "/workspace-bundles/install-proposals/{proposal_id}/connector-bindings" +) +async def bind_bundle_connector( + proposal_id: str, body: BundleConnectorBindingBody +) -> dict: + try: + _installer().bind_connector( + proposal_id, + expected_version=body.expected_version, + slot_id=body.slot_id, + connector_id=body.connector_id, + opaque_connection_id=body.connection_id, + authorized_by=body.actor_id, + ) + return _payload(proposal_id) + except Exception as exc: + raise _error(exc) from exc + + +@router.post( + "/workspace-bundles/install-proposals/{proposal_id}/local-path-bindings" +) +async def bind_bundle_local_path( + proposal_id: str, body: BundleLocalPathBindingBody, request: Request +) -> dict: + try: + _installer().bind_local_path( + proposal_id, + expected_version=body.expected_version, + slot_id=body.slot_id, + local_path=_authorized_local_path(request, body.local_path), + authorized_by=body.actor_id, + ) + return _payload(proposal_id) + except Exception as exc: + raise _error(exc) from exc + + +@router.post( + "/workspace-bundles/install-proposals/{proposal_id}/script-approvals" +) +async def approve_bundle_script( + proposal_id: str, body: BundleScriptApprovalBody +) -> dict: + try: + _installer().approve_script_action( + proposal_id, + expected_version=body.expected_version, + action_id=body.action_id, + authorized_by=body.actor_id, + ) + return _payload(proposal_id) + except Exception as exc: + raise _error(exc) from exc + + +@router.post( + "/workspace-bundles/install-proposals/{proposal_id}/materialize" +) +async def materialize_bundle( + proposal_id: str, + body: BundleMaterializeBody, + authorization: Annotated[str, Header(alias="Authorization")], +) -> dict: + proposal = get_default_run_journal().get_workspace_bundle_install_proposal( + proposal_id + ) + if proposal is None: + return _payload(proposal_id) + binding = get_workspace_resolver().store.get_binding( + body.email, + proposal.space_id, + body.user_id, + ) + if binding is None: + raise HTTPException( + status_code=404, + detail={"code": "workspace_binding_not_found"}, + ) + space_root = Path(binding.workspace_root).expanduser().resolve() + if not space_root.is_dir(): + raise HTTPException( + status_code=409, + detail={"code": "workspace_binding_unavailable"}, + ) + cloud = None + try: + cloud = _cloud(authorization) + await _installer(cloud).materialize( + proposal_id, + expected_version=body.expected_version, + space_root=space_root, + actor_id=body.actor_id, + allow_content_repository_init=( + body.allow_content_repository_init + ), + ) + return _payload(proposal_id) + except Exception as exc: + raise _error(exc) from exc + finally: + if cloud is not None: + await cloud.close() diff --git a/backend/app/router.py b/backend/app/router.py index 26cf7577e..793b8a8ab 100644 --- a/backend/app/router.py +++ b/backend/app/router.py @@ -37,6 +37,7 @@ from app.controller import ( task_controller, tool_controller, workspace_controller, + workspace_bundle_controller, workspace_git_controller, ) @@ -130,6 +131,12 @@ def register_routers(app: FastAPI, prefix: str = "") -> None: "tags": ["workspace"], "description": "Space-level local workspace binding", }, + { + "router": workspace_bundle_controller.router, + "tags": ["Workforce Bundles"], + "description": "Review-first local Bundle installation", + "self_authenticated": True, + }, { "router": workspace_git_controller.router, "tags": ["workspace-git"], diff --git a/backend/app/run_journal/__init__.py b/backend/app/run_journal/__init__.py index 2863a2fbc..d794ffa5b 100644 --- a/backend/app/run_journal/__init__.py +++ b/backend/app/run_journal/__init__.py @@ -47,6 +47,8 @@ from app.run_journal.models import ( ToolCallRecord, WorkspaceConfigMaterializationRecord, WorkspaceConfigRevisionRecord, + WorkspaceBundleInstallProposalRecord, + WorkspaceBundleLocalBindingRecord, WorkspaceOverlayEntryRecord, WorkspaceReadSnapshotRecord, WorkspaceSnapshotRangeRecord, @@ -116,6 +118,8 @@ __all__ = [ "ToolCallRecord", "WorkspaceConfigMaterializationRecord", "WorkspaceConfigRevisionRecord", + "WorkspaceBundleInstallProposalRecord", + "WorkspaceBundleLocalBindingRecord", "WorkspaceOverlayEntryRecord", "WorkspaceReadSnapshotRecord", "WorkspaceSnapshotRangeRecord", diff --git a/backend/app/run_journal/models.py b/backend/app/run_journal/models.py index 0db02e7f4..6cd2f12b5 100644 --- a/backend/app/run_journal/models.py +++ b/backend/app/run_journal/models.py @@ -125,6 +125,41 @@ class WorkspaceConfigMaterializationRecord: updated_at: float +@dataclass(frozen=True) +class WorkspaceBundleInstallProposalRecord: + proposal_id: str + request_id: str + space_id: str + bundle_id: str + revision_id: str + config_placement: str + state: str + version: int + manifest: dict[str, Any] + manifest_digest: str + assets: tuple[dict[str, Any], ...] + install_plan: dict[str, Any] + decided_by: str | None + decided_at: float | None + error_code: str | None + created_at: float + updated_at: float + + +@dataclass(frozen=True) +class WorkspaceBundleLocalBindingRecord: + binding_id: str + proposal_id: str + slot_id: str + binding_kind: str + connector_id: str | None + opaque_connection_id: str | None + local_path: str | None + required_grants: tuple[str, ...] + authorized_by: str + authorized_at: float + + @dataclass(frozen=True) class EffectiveEnvironmentSpecRecord: environment_spec_id: str @@ -482,6 +517,7 @@ class StartupReconciliationResult: outcome_unknown_tool_call_ids: tuple[str, ...] pending_approval_ids: tuple[str, ...] reconcilable_command_ids: tuple[str, ...] + reconcilable_bundle_install_ids: tuple[str, ...] = () @dataclass(frozen=True) diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index 586aba238..6a5bdac7c 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -68,6 +68,8 @@ from app.run_journal.models import ( ToolCallRecord, WorkspaceConfigMaterializationRecord, WorkspaceConfigRevisionRecord, + WorkspaceBundleInstallProposalRecord, + WorkspaceBundleLocalBindingRecord, WorkspaceOverlayEntryRecord, WorkspaceReadSnapshotRecord, WorkspaceSnapshotRangeRecord, @@ -95,7 +97,7 @@ from app.workspace_config.models import ( canonical_json, ) -SCHEMA_VERSION = 13 +SCHEMA_VERSION = 14 logger = logging.getLogger("run_journal") _MIGRATION_V1 = """ @@ -1020,6 +1022,68 @@ PRAGMA user_version = 13; COMMIT; """ +_MIGRATION_V14 = """ +BEGIN IMMEDIATE; + +CREATE TABLE workspace_bundle_install_proposals ( + proposal_id TEXT PRIMARY KEY, + request_id TEXT NOT NULL UNIQUE, + space_id TEXT NOT NULL, + bundle_id TEXT NOT NULL, + revision_id TEXT NOT NULL, + config_placement TEXT NOT NULL CHECK ( + config_placement IN ('in_repo', 'sidecar') + ), + state TEXT NOT NULL CHECK ( + state IN ( + 'proposed', 'approved', 'materializing', 'materialized', + 'rejected', 'needs_attention' + ) + ), + version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0), + manifest_json TEXT NOT NULL, + manifest_digest TEXT NOT NULL CHECK (length(manifest_digest) = 64), + assets_json TEXT NOT NULL, + install_plan_json TEXT NOT NULL, + decided_by TEXT, + decided_at REAL, + error_code TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + CHECK ( + (decided_by IS NULL AND decided_at IS NULL) + OR (decided_by IS NOT NULL AND decided_at IS NOT NULL) + ) +); + +CREATE INDEX workspace_bundle_install_proposals_space_idx +ON workspace_bundle_install_proposals(space_id, updated_at DESC); + +CREATE TABLE workspace_bundle_local_bindings ( + binding_id TEXT PRIMARY KEY, + proposal_id TEXT NOT NULL REFERENCES workspace_bundle_install_proposals( + proposal_id + ) ON DELETE CASCADE, + slot_id TEXT NOT NULL, + binding_kind TEXT NOT NULL CHECK ( + binding_kind IN ('connector', 'local_path', 'script_approval') + ), + connector_id TEXT, + opaque_connection_id TEXT, + local_path TEXT, + required_grants_json TEXT NOT NULL, + authorized_by TEXT NOT NULL, + authorized_at REAL NOT NULL, + UNIQUE(proposal_id, slot_id) +); + +INSERT OR IGNORE INTO run_journal_migrations(version, applied_at) +VALUES (14, CAST(strftime('%s', 'now') AS REAL)); + +PRAGMA user_version = 14; +COMMIT; +""" + class RunJournalError(RuntimeError): """Base error for local RunJournal operations.""" @@ -1352,6 +1416,387 @@ class SQLiteRunJournal: else None ) + def get_latest_workspace_config_materialization( + self, space_id: str + ) -> WorkspaceConfigMaterializationRecord | None: + with self._lock: + row = self._connection.execute( + """ + SELECT * FROM workspace_config_materializations + WHERE space_id = ? AND state = 'materialized' + ORDER BY updated_at DESC, materialization_id DESC + LIMIT 1 + """, + (space_id,), + ).fetchone() + return ( + self._workspace_config_materialization_from_row(row) + if row is not None + else None + ) + + def put_workspace_bundle_install_proposal( + self, + *, + proposal_id: str, + request_id: str, + space_id: str, + bundle_id: str, + revision_id: str, + config_placement: str, + manifest: dict[str, Any], + assets: list[dict[str, Any]], + install_plan: dict[str, Any], + now: float | None = None, + ) -> WorkspaceBundleInstallProposalRecord: + """Persist a reviewable install proposal without granting anything.""" + + if any( + not value.strip() + for value in ( + proposal_id, + request_id, + space_id, + bundle_id, + revision_id, + ) + ): + raise ValueError("Bundle install proposal identity is required") + if config_placement not in {"in_repo", "sidecar"}: + raise ValueError("invalid config_placement") + timestamp = now if now is not None else time.time() + manifest_json = canonical_json(manifest) + manifest_digest = canonical_digest(manifest) + assets_json = canonical_json(assets) + plan_json = canonical_json(install_plan) + expected = ( + proposal_id, + request_id, + space_id, + bundle_id, + revision_id, + config_placement, + manifest_json, + manifest_digest, + assets_json, + plan_json, + ) + with self._write_transaction() as connection: + row = connection.execute( + """ + SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ? OR request_id = ? + """, + (proposal_id, request_id), + ).fetchone() + if row is not None: + actual = ( + row["proposal_id"], + row["request_id"], + row["space_id"], + row["bundle_id"], + row["revision_id"], + row["config_placement"], + row["manifest_json"], + row["manifest_digest"], + row["assets_json"], + row["install_plan_json"], + ) + if actual != expected: + raise IdempotencyConflictError( + "Bundle install request was reused with another payload" + ) + return self._workspace_bundle_install_proposal_from_row(row) + connection.execute( + """ + INSERT INTO workspace_bundle_install_proposals( + proposal_id, request_id, space_id, bundle_id, + revision_id, config_placement, state, version, + manifest_json, manifest_digest, assets_json, + install_plan_json, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, 'proposed', 0, ?, ?, ?, ?, ?, ?) + """, + ( + proposal_id, + request_id, + space_id, + bundle_id, + revision_id, + config_placement, + manifest_json, + manifest_digest, + assets_json, + plan_json, + timestamp, + timestamp, + ), + ) + row = connection.execute( + """SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ?""", + (proposal_id,), + ).fetchone() + assert row is not None + return self._workspace_bundle_install_proposal_from_row(row) + + def get_workspace_bundle_install_proposal( + self, proposal_id: str + ) -> WorkspaceBundleInstallProposalRecord | None: + with self._lock: + row = self._connection.execute( + """SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ?""", + (proposal_id,), + ).fetchone() + return ( + self._workspace_bundle_install_proposal_from_row(row) + if row is not None + else None + ) + + def get_materialized_workspace_bundle_proposal( + self, *, space_id: str, revision_id: str + ) -> WorkspaceBundleInstallProposalRecord | None: + with self._lock: + row = self._connection.execute( + """ + SELECT * FROM workspace_bundle_install_proposals + WHERE space_id = ? AND revision_id = ? + AND state = 'materialized' + ORDER BY updated_at DESC, proposal_id DESC + LIMIT 1 + """, + (space_id, revision_id), + ).fetchone() + return ( + self._workspace_bundle_install_proposal_from_row(row) + if row is not None + else None + ) + + def transition_workspace_bundle_install_proposal( + self, + proposal_id: str, + *, + expected_version: int, + state: str, + decided_by: str | None = None, + error_code: str | None = None, + now: float | None = None, + ) -> WorkspaceBundleInstallProposalRecord: + allowed = { + "proposed": {"approved", "rejected"}, + "approved": {"materializing", "rejected"}, + "materializing": {"materialized", "needs_attention"}, + "needs_attention": {"materializing", "rejected"}, + "materialized": set(), + "rejected": set(), + } + if state not in allowed: + raise ValueError("invalid Bundle install proposal state") + timestamp = now if now is not None else time.time() + with self._write_transaction() as connection: + row = connection.execute( + """SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ?""", + (proposal_id,), + ).fetchone() + if row is None: + raise RunNotFoundError( + f"Bundle install proposal {proposal_id!r} does not exist" + ) + if row["state"] == state: + if state in {"approved", "rejected"} and ( + not decided_by or row["decided_by"] != decided_by + ): + raise IdempotencyConflictError( + "Bundle install decision actor does not match replay" + ) + return self._workspace_bundle_install_proposal_from_row(row) + if int(row["version"]) != expected_version: + raise OptimisticConcurrencyError( + f"Bundle install proposal {proposal_id!r} changed" + ) + if state not in allowed[row["state"]]: + raise InvalidRunTransitionError( + f"Bundle install proposal cannot move from " + f"{row['state']!r} to {state!r}" + ) + decision_actor = row["decided_by"] + decision_at = row["decided_at"] + if state in {"approved", "rejected"}: + if not decided_by or not decided_by.strip(): + raise ValueError("decided_by is required for user decision") + decision_actor = decided_by + decision_at = timestamp + updated = connection.execute( + """ + UPDATE workspace_bundle_install_proposals + SET state = ?, version = version + 1, + decided_by = ?, decided_at = ?, error_code = ?, + updated_at = ? + WHERE proposal_id = ? AND version = ? AND state = ? + """, + ( + state, + decision_actor, + decision_at, + error_code, + timestamp, + proposal_id, + expected_version, + row["state"], + ), + ) + if updated.rowcount != 1: + raise OptimisticConcurrencyError( + f"Bundle install proposal {proposal_id!r} changed" + ) + row = connection.execute( + """SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ?""", + (proposal_id,), + ).fetchone() + assert row is not None + return self._workspace_bundle_install_proposal_from_row(row) + + def put_workspace_bundle_local_binding( + self, + *, + proposal_id: str, + expected_proposal_version: int, + slot_id: str, + binding_kind: str, + connector_id: str | None, + opaque_connection_id: str | None, + local_path: str | None, + required_grants: list[str], + authorized_by: str, + now: float | None = None, + ) -> tuple[ + WorkspaceBundleLocalBindingRecord, + WorkspaceBundleInstallProposalRecord, + ]: + if binding_kind not in {"connector", "local_path", "script_approval"}: + raise ValueError("invalid Bundle local binding kind") + if not slot_id.strip() or not authorized_by.strip(): + raise ValueError("binding slot and authorizer are required") + if binding_kind == "connector" and ( + not connector_id or not opaque_connection_id + ): + raise ValueError("connector binding requires connector and connection ids") + if binding_kind == "local_path" and not local_path: + raise ValueError("local path binding requires a path") + if binding_kind == "script_approval" and any( + value is not None + for value in (connector_id, opaque_connection_id, local_path) + ): + raise ValueError("script approval cannot carry a resource binding") + binding_id = "bundlebind_" + canonical_digest( + {"proposal_id": proposal_id, "slot_id": slot_id} + )[:32] + grants_json = canonical_json(sorted(set(required_grants))) + expected = ( + binding_id, + proposal_id, + slot_id, + binding_kind, + connector_id, + opaque_connection_id, + local_path, + grants_json, + authorized_by, + ) + timestamp = now if now is not None else time.time() + with self._write_transaction() as connection: + proposal = connection.execute( + """SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ?""", + (proposal_id,), + ).fetchone() + if proposal is None: + raise RunNotFoundError( + f"Bundle install proposal {proposal_id!r} does not exist" + ) + row = connection.execute( + """SELECT * FROM workspace_bundle_local_bindings + WHERE proposal_id = ? AND slot_id = ?""", + (proposal_id, slot_id), + ).fetchone() + if row is not None: + actual = ( + row["binding_id"], + row["proposal_id"], + row["slot_id"], + row["binding_kind"], + row["connector_id"], + row["opaque_connection_id"], + row["local_path"], + row["required_grants_json"], + row["authorized_by"], + ) + if actual != expected: + raise IdempotencyConflictError( + f"Bundle slot {slot_id!r} already has another decision" + ) + return ( + self._workspace_bundle_local_binding_from_row(row), + self._workspace_bundle_install_proposal_from_row(proposal), + ) + if int(proposal["version"]) != expected_proposal_version: + raise OptimisticConcurrencyError( + f"Bundle install proposal {proposal_id!r} changed" + ) + if proposal["state"] not in {"approved", "needs_attention"}: + raise InvalidRunTransitionError( + "Bundle resources can only be bound after approval" + ) + connection.execute( + """ + INSERT INTO workspace_bundle_local_bindings( + binding_id, proposal_id, slot_id, binding_kind, + connector_id, opaque_connection_id, local_path, + required_grants_json, authorized_by, authorized_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + (*expected, timestamp), + ) + connection.execute( + """UPDATE workspace_bundle_install_proposals + SET version = version + 1, updated_at = ? + WHERE proposal_id = ? AND version = ?""", + (timestamp, proposal_id, expected_proposal_version), + ) + row = connection.execute( + """SELECT * FROM workspace_bundle_local_bindings + WHERE binding_id = ?""", + (binding_id,), + ).fetchone() + proposal = connection.execute( + """SELECT * FROM workspace_bundle_install_proposals + WHERE proposal_id = ?""", + (proposal_id,), + ).fetchone() + assert row is not None and proposal is not None + return ( + self._workspace_bundle_local_binding_from_row(row), + self._workspace_bundle_install_proposal_from_row(proposal), + ) + + def list_workspace_bundle_local_bindings( + self, proposal_id: str + ) -> tuple[WorkspaceBundleLocalBindingRecord, ...]: + with self._lock: + rows = self._connection.execute( + """SELECT * FROM workspace_bundle_local_bindings + WHERE proposal_id = ? ORDER BY slot_id""", + (proposal_id,), + ).fetchall() + return tuple( + self._workspace_bundle_local_binding_from_row(row) + for row in rows + ) + def transition_workspace_config_revision( self, revision_id: str, @@ -7511,6 +7956,23 @@ class SQLiteRunJournal: ORDER BY updated_at """ ).fetchall() + bundle_installs = connection.execute( + """ + SELECT proposal_id FROM workspace_bundle_install_proposals + WHERE state = 'materializing' + ORDER BY updated_at, proposal_id + """ + ).fetchall() + connection.execute( + """ + UPDATE workspace_bundle_install_proposals + SET state = 'needs_attention', version = version + 1, + error_code = 'desktop_restarted_during_materialization', + updated_at = ? + WHERE state = 'materializing' + """, + (timestamp,), + ) return StartupReconciliationResult( interrupted_run_ids=tuple(interrupted_runs), completed_cancel_run_ids=tuple(completed_cancels), @@ -7523,6 +7985,9 @@ class SQLiteRunJournal: reconcilable_command_ids=tuple( row["command_id"] for row in commands ), + reconcilable_bundle_install_ids=tuple( + row["proposal_id"] for row in bundle_installs + ), ) def persist_remote_command( @@ -8449,6 +8914,8 @@ class SQLiteRunJournal: self._connection.executescript(_MIGRATION_V12) if version < 13: self._connection.executescript(_MIGRATION_V13) + if version < 14: + self._connection.executescript(_MIGRATION_V14) @contextmanager def _write_transaction(self) -> Iterator[sqlite3.Connection]: @@ -8738,6 +9205,51 @@ class SQLiteRunJournal: updated_at=float(row["updated_at"]), ) + @staticmethod + def _workspace_bundle_install_proposal_from_row( + row: sqlite3.Row, + ) -> WorkspaceBundleInstallProposalRecord: + return WorkspaceBundleInstallProposalRecord( + proposal_id=row["proposal_id"], + request_id=row["request_id"], + space_id=row["space_id"], + bundle_id=row["bundle_id"], + revision_id=row["revision_id"], + config_placement=row["config_placement"], + state=row["state"], + version=int(row["version"]), + manifest=json.loads(row["manifest_json"]), + manifest_digest=row["manifest_digest"], + assets=tuple(json.loads(row["assets_json"])), + install_plan=json.loads(row["install_plan_json"]), + decided_by=row["decided_by"], + decided_at=( + float(row["decided_at"]) + if row["decided_at"] is not None + else None + ), + error_code=row["error_code"], + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + ) + + @staticmethod + def _workspace_bundle_local_binding_from_row( + row: sqlite3.Row, + ) -> WorkspaceBundleLocalBindingRecord: + return WorkspaceBundleLocalBindingRecord( + binding_id=row["binding_id"], + proposal_id=row["proposal_id"], + slot_id=row["slot_id"], + binding_kind=row["binding_kind"], + connector_id=row["connector_id"], + opaque_connection_id=row["opaque_connection_id"], + local_path=row["local_path"], + required_grants=tuple(json.loads(row["required_grants_json"])), + authorized_by=row["authorized_by"], + authorized_at=float(row["authorized_at"]), + ) + @staticmethod def _effective_environment_spec_from_row( row: sqlite3.Row, diff --git a/backend/app/workspace_bundle/__init__.py b/backend/app/workspace_bundle/__init__.py new file mode 100644 index 000000000..780b26258 --- /dev/null +++ b/backend/app/workspace_bundle/__init__.py @@ -0,0 +1,19 @@ +from app.workspace_bundle.cloud import ( + HttpWorkspaceBundleCloudTransport, + WorkspaceBundleCloudError, + WorkspaceBundleCloudTransport, +) +from app.workspace_bundle.installer import ( + WorkspaceBundleBindingsIncomplete, + WorkspaceBundleInstallError, + WorkspaceBundleInstaller, +) + +__all__ = [ + "HttpWorkspaceBundleCloudTransport", + "WorkspaceBundleBindingsIncomplete", + "WorkspaceBundleCloudError", + "WorkspaceBundleCloudTransport", + "WorkspaceBundleInstallError", + "WorkspaceBundleInstaller", +] diff --git a/backend/app/workspace_bundle/cloud.py b/backend/app/workspace_bundle/cloud.py new file mode 100644 index 000000000..6f285a8df --- /dev/null +++ b/backend/app/workspace_bundle/cloud.py @@ -0,0 +1,196 @@ +"""Device-authenticated Cloud transport for Workforce Bundle installation.""" + +from __future__ import annotations + +from typing import Any, Protocol +from urllib.parse import urlparse + +import httpx + + +class WorkspaceBundleCloudError(RuntimeError): + def __init__(self, status_code: int, detail: Any) -> None: + self.status_code = status_code + self.detail = detail + super().__init__(f"Workspace Bundle Cloud API returned {status_code}") + + +class WorkspaceBundleCloudTransport(Protocol): + async def get_revision( + self, bundle_id: str, revision_id: str + ) -> dict[str, Any]: ... + + async def install( + self, space_id: str, bundle_id: str, revision_id: str + ) -> dict[str, Any]: ... + + async def get_environment(self, space_id: str) -> dict[str, Any]: ... + + async def upgrade( + self, + space_id: str, + *, + revision_id: str, + expected_installed_revision_id: str, + expected_version: int, + ) -> dict[str, Any]: ... + + async def bind_connection( + self, space_id: str, payload: dict[str, Any] + ) -> dict[str, Any]: ... + + async def download_asset( + self, bundle_id: str, revision_id: str, asset_id: str + ) -> bytes: ... + + async def put_environment_projection( + self, payload: dict[str, Any] + ) -> dict[str, Any]: ... + + async def close(self) -> None: ... + + +class HttpWorkspaceBundleCloudTransport: + MAX_ASSET_BYTES = 16 * 1024 * 1024 + + def __init__( + self, + *, + server_url: str, + authorization: str, + desktop_instance_id: str, + timeout_seconds: float = 30.0, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self.base_url = self._api_base(server_url) + if not authorization.strip(): + raise ValueError("Cloud authorization is required") + if not desktop_instance_id.strip(): + raise ValueError("Desktop instance id is required") + self.headers = { + "Authorization": authorization, + "X-Desktop-Instance-ID": desktop_instance_id, + } + self.client = httpx.AsyncClient( + timeout=timeout_seconds, + transport=transport, + ) + + @staticmethod + def _api_base(server_url: str) -> str: + value = server_url.strip().rstrip("/") + parsed = urlparse(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("server_url must be an absolute HTTP(S) URL") + if value.endswith("/api/v1"): + return value + if value.endswith("/api"): + return f"{value}/v1" + return f"{value}/api/v1" + + async def _json( + self, + method: str, + path: str, + payload: dict[str, Any] | None = None, + ) -> dict[str, Any]: + response = await self.client.request( + method, + f"{self.base_url}{path}", + headers=self.headers, + json=payload, + ) + if response.is_error: + try: + detail: Any = response.json() + except ValueError: + detail = response.text[:2000] + raise WorkspaceBundleCloudError(response.status_code, detail) + result = response.json() + if not isinstance(result, dict): + raise WorkspaceBundleCloudError( + 502, "Workspace Bundle API returned a non-object response" + ) + return result + + async def get_revision( + self, bundle_id: str, revision_id: str + ) -> dict[str, Any]: + return await self._json( + "GET", + f"/workspace-bundles/{bundle_id}/revisions/{revision_id}", + ) + + async def install( + self, space_id: str, bundle_id: str, revision_id: str + ) -> dict[str, Any]: + return await self._json( + "POST", + f"/spaces/{space_id}/bundle:install", + {"bundle_id": bundle_id, "revision_id": revision_id}, + ) + + async def get_environment(self, space_id: str) -> dict[str, Any]: + return await self._json("GET", f"/spaces/{space_id}/environment") + + async def upgrade( + self, + space_id: str, + *, + revision_id: str, + expected_installed_revision_id: str, + expected_version: int, + ) -> dict[str, Any]: + return await self._json( + "POST", + f"/spaces/{space_id}/bundle:upgrade", + { + "revision_id": revision_id, + "expected_installed_revision_id": ( + expected_installed_revision_id + ), + "expected_version": expected_version, + }, + ) + + async def bind_connection( + self, space_id: str, payload: dict[str, Any] + ) -> dict[str, Any]: + return await self._json( + "POST", f"/spaces/{space_id}/bindings", payload + ) + + async def download_asset( + self, bundle_id: str, revision_id: str, asset_id: str + ) -> bytes: + descriptor = await self._json( + "GET", + f"/workspace-bundles/{bundle_id}/revisions/" + f"{revision_id}/assets/{asset_id}:download", + ) + url = descriptor.get("download_url") + if not isinstance(url, str) or not url: + raise WorkspaceBundleCloudError(502, "Asset URL is missing") + total = 0 + chunks: list[bytes] = [] + async with self.client.stream("GET", url) as response: + if response.is_error: + raise WorkspaceBundleCloudError( + response.status_code, "Bundle asset download failed" + ) + async for chunk in response.aiter_bytes(): + total += len(chunk) + if total > self.MAX_ASSET_BYTES: + raise WorkspaceBundleCloudError( + 413, "Bundle asset exceeds Desktop limit" + ) + chunks.append(chunk) + return b"".join(chunks) + + async def put_environment_projection( + self, payload: dict[str, Any] + ) -> dict[str, Any]: + return await self._json("POST", "/sync/environment-specs", payload) + + async def close(self) -> None: + await self.client.aclose() diff --git a/backend/app/workspace_bundle/installer.py b/backend/app/workspace_bundle/installer.py new file mode 100644 index 000000000..dac633421 --- /dev/null +++ b/backend/app/workspace_bundle/installer.py @@ -0,0 +1,531 @@ +"""Review-first local Workforce Bundle installation and materialization.""" + +from __future__ import annotations + +import asyncio +import hashlib +from pathlib import Path +from typing import Any + +from app.run_journal import ( + InvalidRunTransitionError, + SQLiteRunJournal, + WorkspaceBundleInstallProposalRecord, + WorkspaceBundleLocalBindingRecord, +) +from app.workspace_bundle.cloud import WorkspaceBundleCloudTransport +from app.workspace_config import ( + ConfigPlacement, + WorkforceBundleManifest, + canonical_digest, +) +from app.workspace_git import ConfigurationRepositoryService + + +class WorkspaceBundleInstallError(RuntimeError): + pass + + +class WorkspaceBundleBindingsIncomplete(WorkspaceBundleInstallError): + def __init__(self, missing_slots: list[str]) -> None: + self.missing_slots = tuple(sorted(missing_slots)) + super().__init__( + "Bundle installation is missing explicit bindings: " + + ", ".join(self.missing_slots) + ) + + +class WorkspaceBundleInstaller: + MAX_TOTAL_ASSET_BYTES = 128 * 1024 * 1024 + + def __init__( + self, + journal: SQLiteRunJournal, + configuration_repository: ConfigurationRepositoryService, + cloud: WorkspaceBundleCloudTransport | None, + ) -> None: + self.journal = journal + self.configuration_repository = configuration_repository + self.cloud = cloud + + async def propose( + self, + *, + proposal_id: str, + request_id: str, + space_id: str, + bundle_id: str, + revision_id: str, + config_placement: ConfigPlacement, + ) -> WorkspaceBundleInstallProposalRecord: + if self.cloud is None: + raise WorkspaceBundleInstallError("Bundle Cloud transport is unavailable") + revision = await self.cloud.get_revision(bundle_id, revision_id) + if revision.get("status") != "published": + raise WorkspaceBundleInstallError( + "Only published Bundle revisions can be installed" + ) + manifest_value = revision.get("manifest") + if not isinstance(manifest_value, dict): + raise WorkspaceBundleInstallError("Bundle manifest is missing") + manifest = WorkforceBundleManifest.model_validate(manifest_value) + if manifest.metadata.id != bundle_id or manifest.revision_id != revision_id: + raise WorkspaceBundleInstallError("Bundle revision identity mismatch") + if revision.get("manifest_digest") != manifest.digest: + raise WorkspaceBundleInstallError("Bundle manifest digest mismatch") + assets = revision.get("assets", []) + if not isinstance(assets, list): + raise WorkspaceBundleInstallError("Bundle asset manifest is invalid") + normalized_assets = [self._validate_asset_descriptor(item) for item in assets] + logical_paths = [item["logical_path"] for item in normalized_assets] + if len(set(logical_paths)) != len(logical_paths): + raise WorkspaceBundleInstallError( + "Bundle asset logical paths must be unique" + ) + install_plan = self._install_plan(manifest, normalized_assets) + return self.journal.put_workspace_bundle_install_proposal( + proposal_id=proposal_id, + request_id=request_id, + space_id=space_id, + bundle_id=bundle_id, + revision_id=revision_id, + config_placement=config_placement.value, + manifest=manifest.canonical_payload(), + assets=normalized_assets, + install_plan=install_plan, + ) + + def decide( + self, + proposal_id: str, + *, + expected_version: int, + approved: bool, + decided_by: str, + ) -> WorkspaceBundleInstallProposalRecord: + return self.journal.transition_workspace_bundle_install_proposal( + proposal_id, + expected_version=expected_version, + state="approved" if approved else "rejected", + decided_by=decided_by, + ) + + def bind_connector( + self, + proposal_id: str, + *, + expected_version: int, + slot_id: str, + connector_id: str, + opaque_connection_id: str, + authorized_by: str, + ) -> tuple[ + WorkspaceBundleLocalBindingRecord, + WorkspaceBundleInstallProposalRecord, + ]: + proposal = self._proposal(proposal_id) + connector = next( + ( + item + for item in proposal.install_plan["connector_slots"] + if item["slot_id"] == slot_id + ), + None, + ) + if connector is None or connector["connector_id"] != connector_id: + raise WorkspaceBundleInstallError( + "Connector does not match a declared Bundle slot" + ) + return self.journal.put_workspace_bundle_local_binding( + proposal_id=proposal_id, + expected_proposal_version=expected_version, + slot_id=slot_id, + binding_kind="connector", + connector_id=connector_id, + opaque_connection_id=opaque_connection_id, + local_path=None, + required_grants=connector["required_grants"], + authorized_by=authorized_by, + ) + + def bind_local_path( + self, + proposal_id: str, + *, + expected_version: int, + slot_id: str, + local_path: Path, + authorized_by: str, + ) -> tuple[ + WorkspaceBundleLocalBindingRecord, + WorkspaceBundleInstallProposalRecord, + ]: + proposal = self._proposal(proposal_id) + if slot_id not in proposal.install_plan["local_path_slots"]: + raise WorkspaceBundleInstallError( + "Local path does not match a declared Bundle slot" + ) + resolved = local_path.expanduser().resolve() + if not resolved.is_dir(): + raise WorkspaceBundleInstallError("Local path must be a directory") + return self.journal.put_workspace_bundle_local_binding( + proposal_id=proposal_id, + expected_proposal_version=expected_version, + slot_id=slot_id, + binding_kind="local_path", + connector_id=None, + opaque_connection_id=None, + local_path=str(resolved), + required_grants=[], + authorized_by=authorized_by, + ) + + def approve_script_action( + self, + proposal_id: str, + *, + expected_version: int, + action_id: str, + authorized_by: str, + ) -> tuple[ + WorkspaceBundleLocalBindingRecord, + WorkspaceBundleInstallProposalRecord, + ]: + proposal = self._proposal(proposal_id) + if action_id not in proposal.install_plan["script_actions"]: + raise WorkspaceBundleInstallError( + "Script action is not declared by this Bundle" + ) + return self.journal.put_workspace_bundle_local_binding( + proposal_id=proposal_id, + expected_proposal_version=expected_version, + slot_id=action_id, + binding_kind="script_approval", + connector_id=None, + opaque_connection_id=None, + local_path=None, + required_grants=[], + authorized_by=authorized_by, + ) + + async def materialize( + self, + proposal_id: str, + *, + expected_version: int, + space_root: Path, + actor_id: str, + allow_content_repository_init: bool = False, + ) -> WorkspaceBundleInstallProposalRecord: + proposal = self._proposal(proposal_id) + if self.cloud is None: + raise WorkspaceBundleInstallError( + "Bundle Cloud transport is unavailable" + ) + if proposal.state == "materialized": + return proposal + if proposal.version != expected_version: + raise InvalidRunTransitionError("Bundle install proposal changed") + bindings = self.journal.list_workspace_bundle_local_bindings(proposal_id) + self._require_complete_bindings(proposal, bindings) + materializing = self.journal.transition_workspace_bundle_install_proposal( + proposal_id, + expected_version=expected_version, + state="materializing", + ) + try: + ( + cloud_installation, + previous_revision_id, + ) = await self._ensure_cloud_installation( + proposal + ) + cloud_version = int(cloud_installation["version"]) + for binding in bindings: + if binding.binding_kind != "connector": + continue + cloud_installation = await self.cloud.bind_connection( + proposal.space_id, + { + "bundle_id": proposal.bundle_id, + "slot_id": binding.slot_id, + "connector_id": binding.connector_id, + "connection_id": binding.opaque_connection_id, + "expected_installation_version": cloud_version, + }, + ) + cloud_version = int(cloud_installation["version"]) + assets = await self._download_assets(proposal) + manifest = WorkforceBundleManifest.model_validate(proposal.manifest) + lock_payload = { + "apiVersion": "eigent.ai/lock/v1alpha1", + "bundleRevision": proposal.revision_id, + "manifestDigest": proposal.manifest_digest, + "assets": [ + { + "ref": f"bundle://{item['logical_path']}", + "digest": item["content_digest"], + } + for item in proposal.assets + ], + "skills": [], + "mcpPackages": [], + } + await asyncio.to_thread( + self.configuration_repository.bootstrap, + space_id=proposal.space_id, + space_root=space_root, + manifest=manifest, + placement=ConfigPlacement(proposal.config_placement), + created_by=actor_id, + lock_payload=lock_payload, + assets=assets, + expected_previous_revision_id=previous_revision_id, + allow_content_repository_init=allow_content_repository_init, + ) + revision = self.journal.get_workspace_config_revision( + proposal.revision_id + ) + if revision is None: + raise WorkspaceBundleInstallError( + "Materialized Bundle revision was not persisted" + ) + self.journal.transition_workspace_config_revision( + proposal.revision_id, + expected_version=revision.version, + status="published", + ) + projection = self._space_projection(proposal, bindings) + await self.cloud.put_environment_projection( + { + "projection_id": ( + "envspace_" + + canonical_digest( + { + "space_id": proposal.space_id, + "revision_id": proposal.revision_id, + "projection_digest": projection["projection_digest"], + } + )[:40] + ), + "space_id": proposal.space_id, + "owner_type": "space", + "owner_id": proposal.space_id, + "semantic_spec_digest": canonical_digest( + proposal.manifest["spec"] + ), + "redacted_spec": projection["redacted_spec"], + "redaction_schema_version": 1, + "projection_digest": projection["projection_digest"], + "capability_revision": "unresolved-at-space-install", + "installation_version": cloud_version, + } + ) + except Exception: + self.journal.transition_workspace_bundle_install_proposal( + proposal_id, + expected_version=materializing.version, + state="needs_attention", + error_code="bundle_materialization_failed", + ) + raise + return self.journal.transition_workspace_bundle_install_proposal( + proposal_id, + expected_version=materializing.version, + state="materialized", + ) + + async def _ensure_cloud_installation( + self, + proposal: WorkspaceBundleInstallProposalRecord, + ) -> tuple[dict[str, Any], str | None]: + assert self.cloud is not None + environment = await self.cloud.get_environment(proposal.space_id) + installation = environment.get("installation") + if installation is None: + return ( + await self.cloud.install( + proposal.space_id, + proposal.bundle_id, + proposal.revision_id, + ), + None, + ) + if not isinstance(installation, dict): + raise WorkspaceBundleInstallError( + "Cloud returned an invalid Bundle installation" + ) + installed_bundle_id = str(installation.get("bundle_id", "")) + installed_revision_id = str( + installation.get("installed_revision_id", "") + ) + if installed_bundle_id != proposal.bundle_id: + raise WorkspaceBundleInstallError( + "Space already uses a different Workforce Bundle" + ) + if installed_revision_id == proposal.revision_id: + local_materialization = ( + self.journal.get_latest_workspace_config_materialization( + proposal.space_id + ) + ) + previous_revision_id = ( + local_materialization.revision_id + if local_materialization is not None + and local_materialization.revision_id != proposal.revision_id + else None + ) + return installation, previous_revision_id + try: + installed_version = int(installation["version"]) + except (KeyError, TypeError, ValueError) as exc: + raise WorkspaceBundleInstallError( + "Cloud Bundle installation version is invalid" + ) from exc + upgraded = await self.cloud.upgrade( + proposal.space_id, + revision_id=proposal.revision_id, + expected_installed_revision_id=installed_revision_id, + expected_version=installed_version, + ) + return upgraded, installed_revision_id + + def _proposal(self, proposal_id: str) -> WorkspaceBundleInstallProposalRecord: + proposal = self.journal.get_workspace_bundle_install_proposal(proposal_id) + if proposal is None: + raise WorkspaceBundleInstallError("Bundle install proposal not found") + return proposal + + @staticmethod + def _validate_asset_descriptor(value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + raise WorkspaceBundleInstallError("Bundle asset descriptor is invalid") + required = ("id", "logical_path", "content_digest", "media_type", "size_bytes") + if any(value.get(key) is None for key in required): + raise WorkspaceBundleInstallError("Bundle asset descriptor is incomplete") + digest = str(value["content_digest"]) + size = int(value["size_bytes"]) + if len(digest) != 64 or set(digest) - set("0123456789abcdef"): + raise WorkspaceBundleInstallError("Bundle asset digest is invalid") + if size < 0 or size > 16 * 1024 * 1024: + raise WorkspaceBundleInstallError("Bundle asset size is invalid") + return { + "id": str(value["id"]), + "logical_path": str(value["logical_path"]), + "content_digest": digest, + "media_type": str(value["media_type"]), + "size_bytes": size, + "provenance": str(value.get("provenance", "bundle_author")), + } + + @staticmethod + def _install_plan( + manifest: WorkforceBundleManifest, + assets: list[dict[str, Any]], + ) -> dict[str, Any]: + script_actions = [ + f"skill.script.execute:{item.ref}" + for item in manifest.spec.skills + if item.ref.startswith("bundle://") + ] + [ + f"mcp.server.start:{item.id}" + for item in manifest.spec.mcp_servers + ] + return { + "connector_slots": [ + { + "slot_id": item.connection_slot, + "connector_id": item.connector, + "required_grants": list(item.required_grants), + } + for item in manifest.spec.connectors + ], + "local_path_slots": sorted( + { + source.slot + for source in manifest.spec.context + if source.kind == "local_path_slot" and source.slot + } + ), + "script_actions": sorted(script_actions), + "permission_profile": manifest.spec.permissions.profile, + "git_policy": manifest.spec.git.model_dump(by_alias=True, mode="json"), + "asset_count": len(assets), + "asset_bytes": sum(int(item["size_bytes"]) for item in assets), + "automatic_grants": [], + } + + @staticmethod + def _require_complete_bindings( + proposal: WorkspaceBundleInstallProposalRecord, + bindings: tuple[WorkspaceBundleLocalBindingRecord, ...], + ) -> None: + present = {binding.slot_id for binding in bindings} + required = { + item["slot_id"] + for item in proposal.install_plan["connector_slots"] + } + required.update(proposal.install_plan["local_path_slots"]) + required.update(proposal.install_plan["script_actions"]) + missing = sorted(required - present) + if missing: + raise WorkspaceBundleBindingsIncomplete(missing) + + async def _download_assets( + self, proposal: WorkspaceBundleInstallProposalRecord + ) -> dict[str, bytes]: + total = sum(int(item["size_bytes"]) for item in proposal.assets) + if total > self.MAX_TOTAL_ASSET_BYTES: + raise WorkspaceBundleInstallError( + "Bundle assets exceed the Desktop installation limit" + ) + downloaded: dict[str, bytes] = {} + for item in proposal.assets: + content = await self.cloud.download_asset( + proposal.bundle_id, + proposal.revision_id, + item["id"], + ) + if len(content) != int(item["size_bytes"]): + raise WorkspaceBundleInstallError("Bundle asset size mismatch") + if hashlib.sha256(content).hexdigest() != item["content_digest"]: + raise WorkspaceBundleInstallError("Bundle asset digest mismatch") + downloaded[item["logical_path"]] = content + return downloaded + + @staticmethod + def _space_projection( + proposal: WorkspaceBundleInstallProposalRecord, + bindings: tuple[WorkspaceBundleLocalBindingRecord, ...], + ) -> dict[str, Any]: + local_paths = [ + { + "slot_id": item.slot_id, + "root_fingerprint_digest": canonical_digest( + {"slot_id": item.slot_id, "local_path": item.local_path} + ), + } + for item in bindings + if item.binding_kind == "local_path" + ] + connectors = [ + { + "slot_id": item.slot_id, + "connector_id": item.connector_id, + "required_grants": list(item.required_grants), + } + for item in bindings + if item.binding_kind == "connector" + ] + redacted = { + "bundle_revision_id": proposal.revision_id, + "manifest_digest": proposal.manifest_digest, + "context_sources": local_paths, + "connector_bindings": connectors, + "permission_profile": proposal.install_plan["permission_profile"], + "git_policy": proposal.install_plan["git_policy"], + } + return { + "redacted_spec": redacted, + "projection_digest": canonical_digest(redacted), + } diff --git a/backend/app/workspace_config/admission.py b/backend/app/workspace_config/admission.py index 91d0fc56b..74da4f236 100644 --- a/backend/app/workspace_config/admission.py +++ b/backend/app/workspace_config/admission.py @@ -16,7 +16,7 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from typing import Any @@ -31,6 +31,7 @@ from app.workspace_config.models import ( EffectiveEnvironmentSpec, LocalMaterialization, ProviderModelCapability, + ResolvedConnectorBinding, ResolvedContextSource, ThinkingEffort, WorkforceBundleManifest, @@ -174,14 +175,103 @@ class EnvironmentAdmissionService: created_by: str, template: EnvironmentAdmissionTemplate, ) -> EnvironmentAdmissionResult: + installed = self.journal.get_latest_workspace_config_materialization( + space_id + ) + effective_template = template + local_context_sources: list[ResolvedContextSource] = [] + connector_bindings: list[ResolvedConnectorBinding] = [] + if installed is not None: + revision = self.journal.get_workspace_config_revision( + installed.revision_id + ) + if revision is None: + raise ValueError( + "Materialized Workspace Bundle revision is missing" + ) + installed_manifest = WorkforceBundleManifest.model_validate( + revision.manifest + ) + proposal = ( + self.journal.get_materialized_workspace_bundle_proposal( + space_id=space_id, + revision_id=installed.revision_id, + ) + ) + if proposal is not None: + bindings = { + item.slot_id: item + for item in self.journal.list_workspace_bundle_local_bindings( + proposal.proposal_id + ) + } + for source in installed_manifest.spec.context: + if source.kind == "bundle_asset": + local_context_sources.append( + ResolvedContextSource( + id=source.id, + kind=source.kind, + logical_uri=source.path, + ) + ) + elif source.kind == "local_path_slot" and source.slot: + binding = bindings.get(source.slot) + if binding is None or not binding.local_path: + raise ValueError( + f"Workspace Bundle path slot " + f"{source.slot!r} is not bound" + ) + path = Path(binding.local_path).expanduser().resolve() + if not path.is_dir(): + raise ValueError( + f"Workspace Bundle path slot " + f"{source.slot!r} is unavailable" + ) + local_context_sources.append( + ResolvedContextSource( + id=source.id, + kind=source.kind, + slot_id=source.slot, + absolute_path=str(path), + root_fingerprint_digest=canonical_digest( + { + "slot_id": source.slot, + "local_path": str(path), + } + ), + ) + ) + connector_bindings.extend( + ResolvedConnectorBinding( + connector_id=item.connector_id or "", + slot_id=item.slot_id, + local_binding_id=item.opaque_connection_id, + required_grants=item.required_grants, + ) + for item in bindings.values() + if item.binding_kind == "connector" + ) + effective_template = replace( + template, + manifest=installed_manifest, + runtime_capability_manifest={ + **template.runtime_capability_manifest, + "workspace_bundle": { + "revision_id": installed.revision_id, + "config_placement": installed.config_placement, + }, + }, + ) + current_profile = self.journal.get_space_permission_profile(space_id) permission_profile_revision = ( f"space:{space_id}:{current_profile.revision}" if current_profile is not None else None ) - local_materialization = LocalMaterialization( - context_sources=( + if not any(item.id == "workspace_root" for item in local_context_sources): + local_context_sources.insert( + 0, ResolvedContextSource( id="workspace_root", kind="local_path_slot", @@ -191,18 +281,23 @@ class EnvironmentAdmissionService: ), ), ) + local_materialization = LocalMaterialization( + context_sources=tuple(local_context_sources), + connector_bindings=tuple(connector_bindings), ) spec = self.resolver.resolve( - manifest=template.manifest, + manifest=effective_template.manifest, owner_type="run", owner_id=run_id, local_materialization=local_materialization, - provider_capability=template.provider_capability, - thinking_effort_override=template.thinking_effort_requested, + provider_capability=effective_template.provider_capability, + thinking_effort_override=( + effective_template.thinking_effort_requested + ), permission_profile_revision_override=(permission_profile_revision), allow_dynamic_effort_remap=True, runtime_capability_manifest={ - **template.runtime_capability_manifest, + **effective_template.runtime_capability_manifest, "workspace": { "space_id": space_id, "logical_root_slot": "workspace_root", @@ -210,10 +305,10 @@ class EnvironmentAdmissionService: }, ) revision = self.journal.put_workspace_config_revision( - revision_id=template.manifest.revision_id, - bundle_id=template.manifest.metadata.id, - revision_number=template.manifest.metadata.revision, - manifest=template.manifest.canonical_payload(), + revision_id=effective_template.manifest.revision_id, + bundle_id=effective_template.manifest.metadata.id, + revision_number=effective_template.manifest.metadata.revision, + manifest=effective_template.manifest.canonical_payload(), status="validated", created_by=created_by, ) @@ -231,7 +326,7 @@ class EnvironmentAdmissionService: provider_capability_revision=spec.provider_capability_revision, ) return EnvironmentAdmissionResult( - template=template, + template=effective_template, spec=spec, persisted_spec=persisted_spec, revision=revision, diff --git a/backend/app/workspace_git/configuration.py b/backend/app/workspace_git/configuration.py index 413700946..fa7faa934 100644 --- a/backend/app/workspace_git/configuration.py +++ b/backend/app/workspace_git/configuration.py @@ -24,6 +24,7 @@ from collections.abc import Iterator from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path +from pathlib import PurePosixPath from typing import Any import yaml @@ -96,6 +97,8 @@ class ConfigurationRepositoryService: placement: ConfigPlacement, created_by: str, lock_payload: dict[str, Any] | None = None, + assets: dict[str, bytes] | None = None, + expected_previous_revision_id: str | None = None, allow_content_repository_init: bool = False, ) -> ConfigurationRepositoryResult: safe_space_id = self._validate_space_id(space_id) @@ -214,34 +217,131 @@ class ConfigurationRepositoryService: manifest_path = configuration_root / "workspace.yaml" workspace_lock_path = configuration_root / "workspace.lock" - desired_files = { + desired_files: dict[Path, dict[str, Any] | bytes] = { manifest_path: manifest.canonical_payload(), workspace_lock_path: lock, } + for logical_path, content in sorted((assets or {}).items()): + relative = self._validate_asset_path(logical_path) + target = configuration_root / relative + try: + target.resolve(strict=False).relative_to( + configuration_root.resolve() + ) + except ValueError as exc: + raise ConfigurationRepositoryError( + "Bundle asset escapes the configuration repository" + ) from exc + if target in desired_files: + raise ConfigurationRepositoryError( + "Bundle asset conflicts with a reserved configuration file" + ) + desired_files[target] = bytes(content) + obsolete_asset_paths: set[Path] = set() + upgrade_mode = False + if manifest_path.exists() or workspace_lock_path.exists(): + if not manifest_path.is_file() or not workspace_lock_path.is_file(): + raise ConfigurationRepositoryError( + "existing Bundle configuration is incomplete" + ) + try: + current_manifest_payload = yaml.safe_load( + manifest_path.read_text(encoding="utf-8") + ) + current_lock_payload = yaml.safe_load( + workspace_lock_path.read_text(encoding="utf-8") + ) + current_manifest = WorkforceBundleManifest.model_validate( + current_manifest_payload + ) + current_lock = WorkspaceLock.model_validate( + current_lock_payload + ) + except (OSError, yaml.YAMLError, ValidationError) as exc: + raise ConfigurationRepositoryError( + "existing Bundle configuration is invalid" + ) from exc + exact_current = ( + current_manifest.canonical_payload() + == manifest.canonical_payload() + and current_lock.canonical_payload() == lock + ) + if not exact_current: + if expected_previous_revision_id is None: + raise ConfigurationRepositoryError( + "Bundle revision upgrade requires an expected previous revision" + ) + if ( + current_manifest.metadata.id != manifest.metadata.id + or current_manifest.revision_id + != expected_previous_revision_id + or current_lock.bundle_revision + != expected_previous_revision_id + ): + raise ConfigurationRepositoryError( + "existing Bundle revision changed before upgrade" + ) + upgrade_mode = True + for dependency in current_lock.assets: + relative = self._validate_asset_path(dependency.ref) + old_path = configuration_root / relative + if old_path not in desired_files: + obsolete_asset_paths.add(old_path) + managed_paths = tuple( + sorted( + (*desired_files, *obsolete_asset_paths), + key=lambda path: path.as_posix(), + ) + ) path_status = self.git.path_status( repository_root, - tuple(desired_files), + managed_paths, ) - if path_status and not self._recoverable_untracked_files( - repository_root, - desired_files, - path_status, - ): + recoverable = ( + self._recoverable_upgrade_changes( + repository_root, + desired_files, + obsolete_asset_paths, + path_status, + ) + if upgrade_mode + else self._recoverable_untracked_files( + repository_root, + desired_files, + path_status, + ) + ) + if path_status and not recoverable: raise ConfigurationRepositoryError( "configuration repository has uncommitted manifest/lock " "changes; bootstrap will not stage or overwrite them" ) - self._write_new_or_equal_yaml( - manifest_path, - manifest.canonical_payload(), - ) - self._write_new_or_equal_yaml( - workspace_lock_path, - lock, - ) + if upgrade_mode: + self._atomic_write_yaml( + manifest_path, manifest.canonical_payload() + ) + self._atomic_write_yaml(workspace_lock_path, lock) + else: + self._write_new_or_equal_yaml( + manifest_path, + manifest.canonical_payload(), + ) + self._write_new_or_equal_yaml( + workspace_lock_path, + lock, + ) + for path, content in desired_files.items(): + if isinstance(content, bytes): + if upgrade_mode: + self._atomic_write_bytes(path, content) + else: + self._write_new_or_equal_bytes(path, content) + for path in obsolete_asset_paths: + if path.exists(): + path.unlink() commit_oid = self.git.commit_paths( repository_root, - (manifest_path, workspace_lock_path), + managed_paths, message=( f"chore(eigent): configure {manifest.metadata.name} " f"v{manifest.metadata.revision}" @@ -323,6 +423,53 @@ class ConfigurationRepositoryService: if temporary.exists(): temporary.unlink() + @staticmethod + def _atomic_write_bytes(path: Path, content: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + try: + with temporary.open("wb") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + try: + directory_fd = os.open(path.parent, os.O_RDONLY) + except OSError: + return + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + finally: + if temporary.exists(): + temporary.unlink() + + @classmethod + def _atomic_write_yaml(cls, path: Path, payload: dict[str, Any]) -> None: + cls._atomic_write_text( + path, + yaml.safe_dump(payload, allow_unicode=True, sort_keys=False), + ) + + @staticmethod + def _validate_asset_path(value: str) -> Path: + logical = value.removeprefix("bundle://") + path = PurePosixPath(logical) + if ( + not logical + or path.is_absolute() + or ".." in path.parts + or "\\" in logical + or ".git" in path.parts + or path.as_posix() in {"workspace.yaml", "workspace.lock"} + or any(part in {"", "."} for part in path.parts) + ): + raise ConfigurationRepositoryError( + "Bundle asset path must be a safe logical path" + ) + return Path(*path.parts) + @classmethod def _write_new_or_equal_yaml( cls, @@ -350,10 +497,25 @@ class ConfigurationRepositoryService: ), ) + @classmethod + def _write_new_or_equal_bytes(cls, path: Path, content: bytes) -> None: + if path.exists(): + try: + if path.is_file() and path.read_bytes() == content: + return + except OSError as exc: + raise ConfigurationRepositoryError( + f"existing Bundle asset is unreadable: {path}" + ) from exc + raise ConfigurationRepositoryError( + f"refusing to overwrite existing Bundle asset: {path}" + ) + cls._atomic_write_bytes(path, content) + def _recoverable_untracked_files( self, repository_root: Path, - desired_files: dict[Path, dict[str, Any]], + desired_files: dict[Path, dict[str, Any] | bytes], path_status: dict[str, str], ) -> bool: root = repository_root.expanduser().resolve() @@ -364,10 +526,43 @@ class ConfigurationRepositoryService: continue if status != "??" or self.git.is_tracked(repository_root, path): return False - if not self._yaml_matches(path, payload): + if not self._content_matches(path, payload): return False return True + def _recoverable_upgrade_changes( + self, + repository_root: Path, + desired_files: dict[Path, dict[str, Any] | bytes], + obsolete_paths: set[Path], + path_status: dict[str, str], + ) -> bool: + """Recognize only writes/deletes from an interrupted prior upgrade.""" + + root = repository_root.expanduser().resolve() + for relative in path_status: + path = root / relative + if path in obsolete_paths: + if path.exists(): + return False + continue + payload = desired_files.get(path) + if payload is None or not self._content_matches(path, payload): + return False + return True + + @staticmethod + def _content_matches( + path: Path, + payload: dict[str, Any] | bytes, + ) -> bool: + if isinstance(payload, bytes): + try: + return path.is_file() and path.read_bytes() == payload + except OSError: + return False + return ConfigurationRepositoryService._yaml_matches(path, payload) + @staticmethod def _yaml_matches(path: Path, payload: dict[str, Any]) -> bool: if not path.is_file(): diff --git a/backend/tests/app/auth/test_local_control.py b/backend/tests/app/auth/test_local_control.py index 6f888d82b..6898e7f55 100644 --- a/backend/tests/app/auth/test_local_control.py +++ b/backend/tests/app/auth/test_local_control.py @@ -15,6 +15,7 @@ from app.controller import ( chat_controller, remote_command_controller, run_controller, + workspace_bundle_controller, ) from app.run_journal import IdempotencyConflictError, RunNotFoundError @@ -238,6 +239,21 @@ def test_every_mutating_chat_route_declares_the_control_principal(): ), f"{sorted(route.methods)} {route.path} is missing control auth" +def test_every_bundle_install_route_requires_local_control_capability(): + assert workspace_bundle_controller.router.routes + for route in workspace_bundle_controller.router.routes: + assert any( + dependency.call is require_local_control_principal + for dependency in route.dependant.dependencies + ), f"{sorted(route.methods)} {route.path} is missing control auth" + assert "server_url" not in ( + workspace_bundle_controller.BundleProposalBody.model_fields + ) + assert "server_url" not in ( + workspace_bundle_controller.BundleMaterializeBody.model_fields + ) + + def test_command_result_maps_missing_command_to_not_found(monkeypatch): monkeypatch.setenv("EIGENT_RUNTIME", "electron") monkeypatch.setenv("EIGENT_LOCAL_CONTROL_CAPABILITY", "secret-1") diff --git a/backend/tests/app/workspace_bundle/test_installer.py b/backend/tests/app/workspace_bundle/test_installer.py new file mode 100644 index 000000000..5cb59e032 --- /dev/null +++ b/backend/tests/app/workspace_bundle/test_installer.py @@ -0,0 +1,556 @@ +from __future__ import annotations + +import hashlib +import subprocess +from pathlib import Path + +import pytest + +from app.run_journal import InvalidRunTransitionError, SQLiteRunJournal +from app.workspace_bundle import ( + WorkspaceBundleBindingsIncomplete, + WorkspaceBundleInstaller, +) +from app.workspace_config import ( + ConfigPlacement, + WorkforceBundleManifest, + canonical_digest, +) +from app.workspace_git import ConfigurationRepositoryService, GitBackend + + +def _manifest(revision: int = 1) -> dict: + return { + "apiVersion": "eigent.ai/v1alpha1", + "kind": "WorkforceBundle", + "metadata": { + "id": "bundle-research", + "name": "Research Workforce", + "revision": revision, + }, + "spec": { + "instructions": { + "coordinator": "bundle://instructions/coordinator.md" + }, + "context": [ + { + "id": "docs", + "kind": "local_path_slot", + "slot": "docs_folder", + } + ], + "skills": [ + { + "ref": "bundle://skills/research.py", + "assignTo": ["lead"], + } + ], + "connectors": [ + { + "id": "source", + "connector": "github", + "connectionSlot": "github_readonly", + "requiredGrants": ["repository.read"], + } + ], + "mcpServers": [], + "agents": [ + { + "id": "lead", + "role": "coordinator", + "modelProfile": "default", + } + ], + "models": { + "default": { + "modelRef": "provider://default", + "thinkingEffort": "high", + } + }, + "permissions": {"profile": "request_approval", "rules": []}, + "git": { + "enabled": True, + "checkpointPolicy": "user_and_run_terminal", + "agentIsolation": "worktree", + "remotePolicy": "prompt", + }, + }, + } + + +class FakeCloud: + def __init__(self, *, lose_projection_response_once: bool = False): + self.contents = { + "asset-instruction": b"Coordinate the research.\n", + "asset-skill": b"def run():\n return 'ok'\n", + "asset-instruction-v2": b"Coordinate the upgraded research.\n", + "asset-skill-v2": b"def run():\n return 'v2'\n", + } + self.installation_version = 0 + self.installed_bundle_id: str | None = None + self.installed_revision_id: str | None = None + self.binding: tuple[str, str] | None = None + self.projection: dict | None = None + self.projections: dict[str, dict] = {} + self.lose_projection_response_once = lose_projection_response_once + + async def get_revision(self, bundle_id, revision_id): + revision_number = int(str(revision_id).rsplit("@", 1)[1]) + manifest = WorkforceBundleManifest.model_validate( + _manifest(revision_number) + ).canonical_payload() + suffix = "" if revision_number == 1 else f"-v{revision_number}" + return { + "id": revision_id, + "bundle_id": bundle_id, + "status": "published", + "manifest": manifest, + "manifest_digest": canonical_digest(manifest), + "assets": [ + self._asset( + f"asset-instruction{suffix}", + "instructions/coordinator.md", + ), + self._asset(f"asset-skill{suffix}", "skills/research.py"), + ], + } + + def _asset(self, asset_id: str, logical_path: str): + content = self.contents[asset_id] + return { + "id": asset_id, + "logical_path": logical_path, + "content_digest": hashlib.sha256(content).hexdigest(), + "media_type": "text/plain", + "size_bytes": len(content), + "provenance": "bundle_author", + } + + async def install(self, space_id, bundle_id, revision_id): + self.installed_bundle_id = bundle_id + self.installed_revision_id = revision_id + return { + "space_id": space_id, + "bundle_id": bundle_id, + "installed_revision_id": revision_id, + "state": ( + "ready_to_materialize" if self.binding else "pending_bindings" + ), + "version": self.installation_version, + } + + async def get_environment(self, space_id): + if self.installed_revision_id is None: + return {"installation": None, "bindings": {}} + return { + "installation": { + "space_id": space_id, + "bundle_id": self.installed_bundle_id, + "installed_revision_id": self.installed_revision_id, + "state": "materialized" if self.projection else "pending_bindings", + "version": self.installation_version, + }, + "bindings": {}, + } + + async def upgrade( + self, + space_id, + *, + revision_id, + expected_installed_revision_id, + expected_version, + ): + assert self.installed_revision_id == expected_installed_revision_id + assert self.installation_version == expected_version + self.installed_revision_id = revision_id + self.installation_version += 1 + return { + "space_id": space_id, + "bundle_id": self.installed_bundle_id, + "installed_revision_id": revision_id, + "state": "pending_bindings", + "version": self.installation_version, + } + + async def bind_connection(self, space_id, payload): + requested = (payload["slot_id"], payload["connection_id"]) + if self.binding != requested: + self.binding = requested + self.installation_version += 1 + return { + "space_id": space_id, + "state": "ready_to_materialize", + "version": self.installation_version, + } + + async def download_asset(self, bundle_id, revision_id, asset_id): + return self.contents[asset_id] + + async def put_environment_projection(self, payload): + projection_id = payload["projection_id"] + if projection_id not in self.projections: + self.projections[projection_id] = payload + self.projection = payload + self.installation_version += 1 + if self.lose_projection_response_once: + self.lose_projection_response_once = False + raise RuntimeError("response lost after Cloud commit") + assert { + key: value + for key, value in self.projections[projection_id].items() + if key != "installation_version" + } == { + key: value + for key, value in payload.items() + if key != "installation_version" + } + return self.projections[projection_id] + + async def close(self): + return None + + +@pytest.fixture +def installer(tmp_path): + journal = SQLiteRunJournal(tmp_path / "journal.sqlite3") + hooks = tmp_path / "empty-hooks" + hooks.mkdir() + config = ConfigurationRepositoryService( + journal, + state_root=tmp_path / "state", + git_backend=GitBackend(hooks_path=hooks), + ) + cloud = FakeCloud() + value = WorkspaceBundleInstaller(journal, config, cloud) + try: + yield value, journal, cloud, tmp_path + finally: + journal.close() + + +def test_local_review_decision_does_not_require_cloud(tmp_path): + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + manifest = WorkforceBundleManifest.model_validate( + _manifest() + ).canonical_payload() + proposal = journal.put_workspace_bundle_install_proposal( + proposal_id="offline-proposal", + request_id="offline-request", + space_id="space-1", + bundle_id="bundle-research", + revision_id="bundle-research@1", + config_placement="sidecar", + manifest=manifest, + assets=[], + install_plan={ + "connector_slots": [], + "local_path_slots": [], + "script_actions": [], + "permission_profile": "request_approval", + "git_policy": {}, + "automatic_grants": [], + }, + ) + service = WorkspaceBundleInstaller( + journal, + ConfigurationRepositoryService( + journal, + state_root=tmp_path / "state", + git_backend=GitBackend(hooks_path=tmp_path / "hooks"), + ), + cloud=None, + ) + + decided = service.decide( + proposal.proposal_id, + expected_version=proposal.version, + approved=True, + decided_by="user-1", + ) + + assert decided.state == "approved" + + +async def _approved_and_bound(installer, journal, tmp_path): + proposal = await installer.propose( + proposal_id="proposal-1", + request_id="request-1", + space_id="space-1", + bundle_id="bundle-research", + revision_id="bundle-research@1", + config_placement=ConfigPlacement.SIDECAR, + ) + assert proposal.state == "proposed" + assert proposal.install_plan["automatic_grants"] == [] + with pytest.raises(InvalidRunTransitionError): + installer.bind_connector( + proposal.proposal_id, + expected_version=proposal.version, + slot_id="github_readonly", + connector_id="github", + opaque_connection_id="connection-1", + authorized_by="user-1", + ) + proposal = installer.decide( + proposal.proposal_id, + expected_version=proposal.version, + approved=True, + decided_by="user-1", + ) + decision_replay = installer.decide( + proposal.proposal_id, + expected_version=0, + approved=True, + decided_by="user-1", + ) + assert decision_replay == proposal + with pytest.raises(WorkspaceBundleBindingsIncomplete) as missing: + await installer.materialize( + proposal.proposal_id, + expected_version=proposal.version, + space_root=tmp_path, + actor_id="user-1", + ) + assert set(missing.value.missing_slots) == { + "docs_folder", + "github_readonly", + "skill.script.execute:bundle://skills/research.py", + } + connector_expected_version = proposal.version + connector_binding, proposal = installer.bind_connector( + proposal.proposal_id, + expected_version=proposal.version, + slot_id="github_readonly", + connector_id="github", + opaque_connection_id="connection-1", + authorized_by="user-1", + ) + replay_binding, replay_proposal = installer.bind_connector( + proposal.proposal_id, + expected_version=connector_expected_version, + slot_id="github_readonly", + connector_id="github", + opaque_connection_id="connection-1", + authorized_by="user-1", + ) + assert replay_binding == connector_binding + assert replay_proposal == proposal + docs = tmp_path / "user-selected-docs" + docs.mkdir() + _, proposal = installer.bind_local_path( + proposal.proposal_id, + expected_version=proposal.version, + slot_id="docs_folder", + local_path=docs, + authorized_by="user-1", + ) + _, proposal = installer.approve_script_action( + proposal.proposal_id, + expected_version=proposal.version, + action_id="skill.script.execute:bundle://skills/research.py", + authorized_by="user-1", + ) + assert len(journal.list_workspace_bundle_local_bindings("proposal-1")) == 3 + return proposal + + +@pytest.mark.asyncio +async def test_install_is_review_first_and_materializes_verified_assets(installer): + service, journal, cloud, tmp_path = installer + proposal = await _approved_and_bound(service, journal, tmp_path) + + result = await service.materialize( + proposal.proposal_id, + expected_version=proposal.version, + space_root=tmp_path, + actor_id="user-1", + ) + replay = await service.materialize( + proposal.proposal_id, + expected_version=proposal.version, + space_root=tmp_path, + actor_id="user-1", + ) + + assert result.state == "materialized" + assert replay == result + assert cloud.binding == ("github_readonly", "connection-1") + assert cloud.projection is not None + serialized_projection = str(cloud.projection) + assert "connection-1" not in serialized_projection + assert str(tmp_path) not in serialized_projection + config_root = tmp_path / "state/spaces/space-1/configuration" + assert (config_root / "instructions/coordinator.md").is_file() + assert (config_root / "skills/research.py").is_file() + paths = subprocess.run( + ("git", "-C", str(config_root), "show", "--name-only", "--format="), + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + assert set(paths) == { + "instructions/coordinator.md", + "skills/research.py", + "workspace.lock", + "workspace.yaml", + } + + +@pytest.mark.asyncio +async def test_projection_response_loss_retries_without_duplicate_grants(tmp_path): + journal = SQLiteRunJournal(tmp_path / "journal.sqlite3") + hooks = tmp_path / "empty-hooks" + hooks.mkdir() + cloud = FakeCloud(lose_projection_response_once=True) + service = WorkspaceBundleInstaller( + journal, + ConfigurationRepositoryService( + journal, + state_root=tmp_path / "state", + git_backend=GitBackend(hooks_path=hooks), + ), + cloud, + ) + try: + proposal = await _approved_and_bound(service, journal, tmp_path) + with pytest.raises(RuntimeError, match="response lost"): + await service.materialize( + proposal.proposal_id, + expected_version=proposal.version, + space_root=tmp_path, + actor_id="user-1", + ) + failed = journal.get_workspace_bundle_install_proposal("proposal-1") + assert failed is not None and failed.state == "needs_attention" + installed = await service.materialize( + proposal.proposal_id, + expected_version=failed.version, + space_root=tmp_path, + actor_id="user-1", + ) + assert installed.state == "materialized" + assert cloud.installation_version == 2 + finally: + journal.close() + + +@pytest.mark.asyncio +async def test_upgrade_reviews_again_and_commits_new_configuration(installer): + service, journal, cloud, tmp_path = installer + first = await _approved_and_bound(service, journal, tmp_path) + first = await service.materialize( + first.proposal_id, + expected_version=first.version, + space_root=tmp_path, + actor_id="user-1", + ) + assert first.state == "materialized" + + second = await service.propose( + proposal_id="proposal-2", + request_id="request-2", + space_id="space-1", + bundle_id="bundle-research", + revision_id="bundle-research@2", + config_placement=ConfigPlacement.SIDECAR, + ) + assert second.state == "proposed" + second = service.decide( + second.proposal_id, + expected_version=second.version, + approved=True, + decided_by="user-1", + ) + _, second = service.bind_connector( + second.proposal_id, + expected_version=second.version, + slot_id="github_readonly", + connector_id="github", + opaque_connection_id="connection-1", + authorized_by="user-1", + ) + docs = tmp_path / "user-selected-docs" + _, second = service.bind_local_path( + second.proposal_id, + expected_version=second.version, + slot_id="docs_folder", + local_path=docs, + authorized_by="user-1", + ) + _, second = service.approve_script_action( + second.proposal_id, + expected_version=second.version, + action_id="skill.script.execute:bundle://skills/research.py", + authorized_by="user-1", + ) + second = await service.materialize( + second.proposal_id, + expected_version=second.version, + space_root=tmp_path, + actor_id="user-1", + ) + + assert second.state == "materialized" + assert cloud.installed_revision_id == "bundle-research@2" + assert len(cloud.projections) == 2 + config_root = tmp_path / "state/spaces/space-1/configuration" + assert (config_root / "instructions/coordinator.md").read_bytes() == ( + b"Coordinate the upgraded research.\n" + ) + assert int( + subprocess.run( + ("git", "-C", str(config_root), "rev-list", "--count", "HEAD"), + check=True, + capture_output=True, + text=True, + ).stdout.strip() + ) == 2 + + +def test_startup_reconciliation_exposes_interrupted_materialization(tmp_path): + path = tmp_path / "journal.sqlite3" + with SQLiteRunJournal(path) as journal: + proposal = journal.put_workspace_bundle_install_proposal( + proposal_id="proposal-crash", + request_id="request-crash", + space_id="space-1", + bundle_id="bundle-research", + revision_id="bundle-research@1", + config_placement="sidecar", + manifest=WorkforceBundleManifest.model_validate( + _manifest() + ).canonical_payload(), + assets=[], + install_plan={ + "connector_slots": [], + "local_path_slots": [], + "script_actions": [], + "permission_profile": "request_approval", + "git_policy": {}, + "automatic_grants": [], + }, + ) + proposal = journal.transition_workspace_bundle_install_proposal( + proposal.proposal_id, + expected_version=proposal.version, + state="approved", + decided_by="user-1", + ) + journal.transition_workspace_bundle_install_proposal( + proposal.proposal_id, + expected_version=proposal.version, + state="materializing", + ) + + with SQLiteRunJournal(path) as reopened: + result = reopened.reconcile_startup(now=10) + proposal = reopened.get_workspace_bundle_install_proposal( + "proposal-crash" + ) + + assert result.reconcilable_bundle_install_ids == ("proposal-crash",) + assert proposal is not None and proposal.state == "needs_attention" + assert proposal.error_code == ( + "desktop_restarted_during_materialization" + ) diff --git a/backend/tests/app/workspace_config/test_admission.py b/backend/tests/app/workspace_config/test_admission.py index 2db9a6b8d..bc15f717e 100644 --- a/backend/tests/app/workspace_config/test_admission.py +++ b/backend/tests/app/workspace_config/test_admission.py @@ -3,7 +3,7 @@ from __future__ import annotations import json from app.run_journal import SQLiteRunJournal -from app.workspace_config import ThinkingEffort +from app.workspace_config import ThinkingEffort, WorkforceBundleManifest from app.workspace_config.admission import ( EnvironmentAdmissionService, LegacyEnvironmentImporter, @@ -147,3 +147,157 @@ def test_admission_pins_current_space_permission_profile(tmp_path): assert result.binding.permission_profile_revision == ( f"space:space-1:{profile.revision}" ) + + +def test_materialized_bundle_replaces_legacy_template_for_new_run(tmp_path): + manifest = WorkforceBundleManifest.model_validate({ + "apiVersion": "eigent.ai/v1alpha1", + "kind": "WorkforceBundle", + "metadata": { + "id": "bundle-team", + "name": "Team Workspace", + "revision": 1, + }, + "spec": { + "context": [ + { + "id": "team_docs", + "kind": "local_path_slot", + "slot": "docs_folder", + } + ], + "connectors": [ + { + "id": "source", + "connector": "github", + "connectionSlot": "github_readonly", + "requiredGrants": ["repository.read"], + } + ], + "models": { + "default": { + "modelRef": "provider://default", + "thinkingEffort": "high", + } + }, + }, + }).canonical_payload() + docs = tmp_path / "team-docs" + docs.mkdir() + with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal: + journal.ensure_run( + run_id="run-installed", + project_id="project-1", + status="pending", + ) + revision = journal.put_workspace_config_revision( + revision_id="bundle-team@1", + bundle_id="bundle-team", + revision_number=1, + manifest=manifest, + status="published", + created_by="user-1", + ) + journal.put_workspace_config_materialization( + materialization_id="materialization-1", + space_id="space-1", + revision_id=revision.revision_id, + config_placement="sidecar", + ) + proposal = journal.put_workspace_bundle_install_proposal( + proposal_id="proposal-1", + request_id="install-1", + space_id="space-1", + bundle_id="bundle-team", + revision_id="bundle-team@1", + config_placement="sidecar", + manifest=manifest, + assets=[], + install_plan={ + "connector_slots": [ + { + "slot_id": "github_readonly", + "connector_id": "github", + "required_grants": ["repository.read"], + } + ], + "local_path_slots": ["docs_folder"], + "script_actions": [], + "permission_profile": "request_approval", + "git_policy": {}, + "automatic_grants": [], + }, + ) + proposal = journal.transition_workspace_bundle_install_proposal( + proposal.proposal_id, + expected_version=proposal.version, + state="approved", + decided_by="user-1", + ) + _, proposal = journal.put_workspace_bundle_local_binding( + proposal_id=proposal.proposal_id, + expected_proposal_version=proposal.version, + slot_id="github_readonly", + binding_kind="connector", + connector_id="github", + opaque_connection_id="private-connection-id", + local_path=None, + required_grants=["repository.read"], + authorized_by="user-1", + ) + _, proposal = journal.put_workspace_bundle_local_binding( + proposal_id=proposal.proposal_id, + expected_proposal_version=proposal.version, + slot_id="docs_folder", + binding_kind="local_path", + connector_id=None, + opaque_connection_id=None, + local_path=str(docs), + required_grants=[], + authorized_by="user-1", + ) + proposal = journal.transition_workspace_bundle_install_proposal( + proposal.proposal_id, + expected_version=proposal.version, + state="materializing", + ) + journal.transition_workspace_bundle_install_proposal( + proposal.proposal_id, + expected_version=proposal.version, + state="materialized", + ) + + legacy = LegacyEnvironmentImporter().build_template( + model_platform="openai", + model_type="gpt-5.5-codex", + auth_source="codex_subscription", + requested_effort=ThinkingEffort.HIGH, + allow_local_system=True, + ) + result = EnvironmentAdmissionService(journal).persist_for_run( + run_id="run-installed", + space_id="space-1", + working_directory=tmp_path, + created_by="user-1", + template=legacy, + ) + + assert result.template.manifest.metadata.id == "bundle-team" + assert result.binding.bundle_revision_id == "bundle-team@1" + local = result.spec.local_materialization + assert next( + item.absolute_path + for item in local.context_sources + if item.id == "team_docs" + ) == str(docs.resolve()) + assert local.connector_bindings[0].local_binding_id == ( + "private-connection-id" + ) + event = next( + item + for item in journal.list_events("run-installed") + if item.event_type == "run.environment_resolved" + ) + event_json = json.dumps(event.payload) + assert "private-connection-id" not in event_json + assert str(docs) not in event_json diff --git a/backend/tests/app/workspace_git/test_configuration_repository.py b/backend/tests/app/workspace_git/test_configuration_repository.py index 6f10ee1cb..bdb1a9c41 100644 --- a/backend/tests/app/workspace_git/test_configuration_repository.py +++ b/backend/tests/app/workspace_git/test_configuration_repository.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import subprocess from pathlib import Path @@ -122,6 +123,55 @@ def test_sidecar_bootstrap_never_mutates_user_space( assert first.materialization.config_placement == "sidecar" +def test_sidecar_bootstrap_commits_verified_bundle_assets(tmp_path, journal): + space = tmp_path / "user-space" + space.mkdir() + service, backend = _service(tmp_path, journal) + + result = service.bootstrap( + space_id="space-1", + space_root=space, + manifest=parse_workforce_manifest(MANIFEST), + placement=ConfigPlacement.SIDECAR, + created_by="user-1", + assets={ + "instructions/coordinator.md": b"Coordinate safely.\n", + "images/icon.bin": b"\x00\x01\x02", + }, + lock_payload={ + "apiVersion": "eigent.ai/lock/v1alpha1", + "bundleRevision": "bundle_local@1", + "manifestDigest": parse_workforce_manifest(MANIFEST).digest, + "assets": [ + { + "ref": "bundle://instructions/coordinator.md", + "digest": hashlib.sha256( + b"Coordinate safely.\n" + ).hexdigest(), + }, + { + "ref": "bundle://images/icon.bin", + "digest": hashlib.sha256( + b"\x00\x01\x02" + ).hexdigest(), + }, + ], + "skills": [], + "mcpPackages": [], + }, + ) + + assert ( + result.configuration_repository_root / "images/icon.bin" + ).read_bytes() == b"\x00\x01\x02" + assert set(backend.show_commit_paths(result.configuration_repository_root)) == { + "images/icon.bin", + "instructions/coordinator.md", + "workspace.lock", + "workspace.yaml", + } + + def test_same_bundle_revision_materializes_into_multiple_spaces( tmp_path, journal, @@ -444,6 +494,100 @@ def test_bootstrap_recovers_exact_untracked_files_after_crash( ) +def test_bundle_upgrade_preserves_user_edits_and_removes_only_clean_old_assets( + tmp_path, + journal, +): + space = tmp_path / "space" + space.mkdir() + service, backend = _service(tmp_path, journal) + first_manifest = parse_workforce_manifest(MANIFEST) + first_assets = { + "instructions/coordinator.md": b"version one\n", + "assets/obsolete.txt": b"remove on upgrade\n", + } + first = service.bootstrap( + space_id="space-1", + space_root=space, + manifest=first_manifest, + placement=ConfigPlacement.SIDECAR, + created_by="user-1", + assets=first_assets, + lock_payload={ + "apiVersion": "eigent.ai/lock/v1alpha1", + "bundleRevision": first_manifest.revision_id, + "manifestDigest": first_manifest.digest, + "assets": [ + { + "ref": f"bundle://{path}", + "digest": hashlib.sha256(content).hexdigest(), + } + for path, content in first_assets.items() + ], + "skills": [], + "mcpPackages": [], + }, + ) + edited_path = first.configuration_repository_root / "instructions/coordinator.md" + edited_path.write_text("user edit\n", encoding="utf-8") + second_manifest = parse_workforce_manifest( + MANIFEST.replace("revision: 1", "revision: 2") + ) + second_assets = {"instructions/coordinator.md": b"version two\n"} + second_lock = { + "apiVersion": "eigent.ai/lock/v1alpha1", + "bundleRevision": second_manifest.revision_id, + "manifestDigest": second_manifest.digest, + "assets": [ + { + "ref": "bundle://instructions/coordinator.md", + "digest": hashlib.sha256(second_assets["instructions/coordinator.md"]).hexdigest(), + } + ], + "skills": [], + "mcpPackages": [], + } + + with pytest.raises(ConfigurationRepositoryError, match="uncommitted"): + service.bootstrap( + space_id="space-1", + space_root=space, + manifest=second_manifest, + placement=ConfigPlacement.SIDECAR, + created_by="user-1", + assets=second_assets, + lock_payload=second_lock, + expected_previous_revision_id=first_manifest.revision_id, + ) + assert edited_path.read_text(encoding="utf-8") == "user edit\n" + + edited_path.write_bytes(first_assets["instructions/coordinator.md"]) + upgraded = service.bootstrap( + space_id="space-1", + space_root=space, + manifest=second_manifest, + placement=ConfigPlacement.SIDECAR, + created_by="user-1", + assets=second_assets, + lock_payload=second_lock, + expected_previous_revision_id=first_manifest.revision_id, + ) + + assert edited_path.read_bytes() == b"version two\n" + assert not ( + upgraded.configuration_repository_root / "assets/obsolete.txt" + ).exists() + assert backend.changed_paths( + upgraded.configuration_repository_root, + ( + upgraded.manifest_path, + upgraded.lock_path, + edited_path, + ), + ) == () + assert int(_git(upgraded.configuration_repository_root, "rev-list", "--count", "HEAD")) == 2 + + def test_revision_conflict_is_detected_before_git_mutation( tmp_path, journal,