vrr/eigent
2
0
Fork 0
mirror of https://github.com/eigent-ai/eigent.git synced 2026-08-24 16:13:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat: add review-first workforce bundle sharing

Browse source
This commit is contained in:
4pmtong 2026-08-07 23:01:53 +08:00
parent c4dce97dff
commit 238f8328f0
14 changed files with 2833 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

336
backend/app/controller/workspace_bundle_controller.py Normal file
View file

@ -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()

7
backend/app/router.py
View file

@ -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"],

4
backend/app/run_journal/__init__.py
View file

@ -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",

36
backend/app/run_journal/models.py
View file

@ -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)

514
backend/app/run_journal/store.py
View file

@ -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,

19
backend/app/workspace_bundle/__init__.py Normal file
View file

@ -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",
]

196
backend/app/workspace_bundle/cloud.py Normal file
View file

@ -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()

531
backend/app/workspace_bundle/installer.py Normal file
View file

@ -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),
}

119
backend/app/workspace_config/admission.py
View file

@ -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,

231
backend/app/workspace_git/configuration.py
View file

@ -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():

16
backend/tests/app/auth/test_local_control.py
View file

@ -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")

556
backend/tests/app/workspace_bundle/test_installer.py Normal file
View file

@ -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"
)

156
backend/tests/app/workspace_config/test_admission.py
View file

@ -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

144
backend/tests/app/workspace_git/test_configuration_repository.py
View file

@ -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,