diff --git a/backend/app/controller/workspace_git_controller.py b/backend/app/controller/workspace_git_controller.py new file mode 100644 index 000000000..6218c999a --- /dev/null +++ b/backend/app/controller/workspace_git_controller.py @@ -0,0 +1,433 @@ +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +"""Authenticated Desktop-local Content Repository API.""" + +from __future__ import annotations + +from pathlib import Path, PurePosixPath +from typing import Annotated, Literal + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field, field_validator + +from app.auth import require_local_control_principal +from app.run_journal import ( + IdempotencyConflictError, + InvalidRunTransitionError, + default_run_journal_path, + get_default_run_journal, +) +from app.utils.workspace_resolver import get_workspace_resolver +from app.workspace_git import ( + ContentRepositoryConsentRequired, + ContentRepositoryError, + ContentRepositoryService, + GitBackendError, + NestedRepositoryError, + NoCheckpointChangesError, + RepositoryStateChangedError, +) +from app.workspace_git.backend import RepositoryDiagnostics + +router = APIRouter(dependencies=[Depends(require_local_control_principal)]) + + +class GitBootstrapBody(BaseModel): + email: str = Field(min_length=1) + user_id: str | int | None = None + allow_init: bool = False + eigent_owned_space: bool = False + + +class GitCheckpointBody(BaseModel): + email: str = Field(min_length=1) + user_id: str | int | None = None + operation_request_id: str = Field(min_length=1, max_length=128) + expected_repo_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$") + paths: list[str] = Field(min_length=1, max_length=500) + path_sources: dict[str, str] + target_role: Literal["user", "project", "run", "agent"] + target_id: str = Field(min_length=1, max_length=256) + actor_id: str = Field(min_length=1, max_length=200) + trigger: str = Field(min_length=1, max_length=200) + message: str = Field(min_length=1, max_length=500) + + @field_validator("paths") + @classmethod + def validate_relative_paths(cls, paths: list[str]) -> list[str]: + normalized: list[str] = [] + for value in paths: + path = PurePosixPath(value) + if ( + not value + or path.is_absolute() + or ".." in path.parts + or value.startswith(("~/", "\\\\")) + or (len(value) > 1 and value[1] == ":") + ): + raise ValueError("Git checkpoint paths must be relative") + normalized.append(path.as_posix()) + if len(set(normalized)) != len(normalized): + raise ValueError("Git checkpoint paths must be unique") + return normalized + + +class GitRestoreBody(BaseModel): + email: str = Field(min_length=1) + user_id: str | int | None = None + checkpoint_id: str = Field( + min_length=1, + max_length=128, + pattern=r"^checkpoint_[0-9a-f]{32}$", + ) + operation_request_id: str = Field(min_length=1, max_length=128) + expected_repo_state_digest: str = Field(pattern=r"^[0-9a-f]{64}$") + + +def _service() -> ContentRepositoryService: + return ContentRepositoryService( + get_default_run_journal(), + state_root=default_run_journal_path().parent / "workspace-git", + ) + + +def _binding_root( + *, + space_id: str, + email: str, + user_id: str | int | None, +) -> Path: + binding = get_workspace_resolver().store.get_binding( + email, + space_id, + user_id, + ) + if binding is None: + raise HTTPException( + status_code=404, + detail={ + "code": "workspace_binding_not_found", + "message": "The Space has no local workspace binding.", + }, + ) + root = Path(binding.workspace_root).expanduser() + if not root.is_dir(): + raise HTTPException( + status_code=409, + detail={ + "code": "workspace_binding_unavailable", + "message": "The bound workspace folder is unavailable.", + }, + ) + return root.resolve() + + +def _assert_repository_binding(repository, root: Path) -> None: + if Path(repository.root_path).expanduser().resolve() != root: + raise HTTPException( + status_code=409, + detail={ + "code": "git_repository_binding_mismatch", + "message": ( + "The persisted Content Repository no longer matches the " + "Space binding. Reconciliation is required." + ), + }, + ) + + +def _git_error(exc: Exception) -> HTTPException: + if isinstance(exc, HTTPException): + return exc + if isinstance(exc, ContentRepositoryConsentRequired): + return HTTPException( + status_code=409, + detail={ + "code": "git_init_consent_required", + "message": str(exc), + }, + ) + if isinstance(exc, RepositoryStateChangedError): + return HTTPException( + status_code=409, + detail={"code": "repo_state_changed", "message": str(exc)}, + ) + if isinstance(exc, NoCheckpointChangesError): + return HTTPException( + status_code=409, + detail={"code": "git_no_changes", "message": str(exc)}, + ) + if isinstance( + exc, + (IdempotencyConflictError, InvalidRunTransitionError), + ): + return HTTPException( + status_code=409, + detail={"code": "git_operation_conflict", "message": str(exc)}, + ) + if isinstance(exc, NestedRepositoryError): + return HTTPException( + status_code=409, + detail={ + "code": "nested_repository_requires_binding", + "message": str(exc), + }, + ) + if isinstance(exc, (ValueError, GitBackendError)): + return HTTPException( + status_code=422, + detail={"code": "invalid_git_operation", "message": str(exc)}, + ) + if isinstance(exc, ContentRepositoryError): + return HTTPException( + status_code=409, + detail={"code": "git_needs_attention", "message": str(exc)}, + ) + return HTTPException( + status_code=500, + detail={"code": "git_operation_failed", "message": "Git failed"}, + ) + + +def _diagnostics_payload(value: RepositoryDiagnostics) -> dict: + return { + "healthy": value.healthy, + "issues": list(value.issues), + "has_submodules": value.has_submodules, + "has_remotes": value.has_remotes, + "repo_state": { + "head_oid": value.state_token.head_oid, + "branch_or_detached_head": ( + value.state_token.branch_or_detached_head + ), + "index_digest": value.state_token.index_digest, + "operation_state": value.state_token.operation_state, + "digest": value.state_token.digest, + }, + } + + +@router.get("/spaces/{space_id}/git/status") +async def git_status( + space_id: str, + email: str = Query(..., min_length=1), + user_id: str | None = Query(None), +): + root = _binding_root(space_id=space_id, email=email, user_id=user_id) + service = _service() + repository = service.journal.get_space_git_repository(space_id=space_id) + try: + if repository is None: + inspection = service.inspect(root) + return { + "space_id": space_id, + "enabled": False, + "enablement": inspection.enablement, + "consent_required": inspection.consent_required, + "existing_repository": inspection.probe.is_repository, + "nested_in_parent": inspection.probe.nested_in_parent, + "diagnostics": ( + _diagnostics_payload(inspection.diagnostics) + if inspection.diagnostics is not None + else None + ), + } + _assert_repository_binding(repository, root) + status = service.status(repository.repository_id) + return { + "space_id": space_id, + "enabled": True, + "repository_id": repository.repository_id, + "state": repository.state, + "ownership": repository.ownership, + "version_coverage": repository.version_coverage, + "hooks_mode": repository.hooks_mode, + "managed_paths": list(status.managed_paths), + "diagnostics": _diagnostics_payload(status.diagnostics), + } + except Exception as exc: + raise _git_error(exc) from exc + + +@router.post("/spaces/{space_id}/git/bootstrap") +async def git_bootstrap(space_id: str, body: GitBootstrapBody): + root = _binding_root( + space_id=space_id, + email=body.email, + user_id=body.user_id, + ) + try: + result = _service().bootstrap( + space_id=space_id, + space_root=root, + allow_init=body.allow_init, + eigent_owned_space=body.eigent_owned_space, + ) + except Exception as exc: + raise _git_error(exc) from exc + return { + "space_id": space_id, + "repository_id": result.repository.repository_id, + "initialized": result.initialized, + "ownership": result.repository.ownership, + "state": result.repository.state, + "version_coverage": result.repository.version_coverage, + "diagnostics": _diagnostics_payload(result.diagnostics), + } + + +@router.get("/spaces/{space_id}/git/diff") +async def git_diff( + space_id: str, + paths: Annotated[list[str], Query(min_length=1, max_length=500)], + source_commit: str | None = Query(None), + email: str = Query(..., min_length=1), + user_id: str | None = Query(None), +): + bound_root = _binding_root( + space_id=space_id, + email=email, + user_id=user_id, + ) + service = _service() + repository = service.journal.get_space_git_repository(space_id=space_id) + if repository is None: + raise HTTPException(status_code=404, detail="Git is not enabled") + _assert_repository_binding(repository, bound_root) + try: + root = Path(repository.root_path) + diff = service.diff( + repository.repository_id, + paths=tuple(root / path for path in paths), + source_commit=source_commit, + ) + return {"repository_id": repository.repository_id, "diff": diff} + except Exception as exc: + raise _git_error(exc) from exc + + +@router.get("/spaces/{space_id}/git/checkpoints") +async def git_checkpoints( + space_id: str, + limit: int = Query(100, ge=1, le=500), + email: str = Query(..., min_length=1), + user_id: str | None = Query(None), +): + bound_root = _binding_root( + space_id=space_id, + email=email, + user_id=user_id, + ) + service = _service() + repository = service.journal.get_space_git_repository(space_id=space_id) + if repository is None: + raise HTTPException(status_code=404, detail="Git is not enabled") + _assert_repository_binding(repository, bound_root) + checkpoints = service.journal.list_git_checkpoints( + repository.repository_id, + limit=limit, + ) + return { + "repository_id": repository.repository_id, + "checkpoints": [ + { + "checkpoint_id": item.checkpoint_id, + "target_role": item.target_role, + "target_id": item.target_id, + "commit_oid": item.commit_oid, + "parent_oid": item.parent_oid, + "paths": list(item.paths), + "actor_id": item.actor_id, + "trigger": item.trigger, + "message": item.message, + "created_at": item.created_at, + } + for item in checkpoints + ], + } + + +@router.post("/spaces/{space_id}/git/checkpoints", status_code=201) +async def git_checkpoint(space_id: str, body: GitCheckpointBody): + root = _binding_root( + space_id=space_id, + email=body.email, + user_id=body.user_id, + ) + service = _service() + repository = service.journal.get_space_git_repository(space_id=space_id) + if repository is None: + raise HTTPException(status_code=404, detail="Git is not enabled") + _assert_repository_binding(repository, root) + try: + checkpoint = service.checkpoint( + repository.repository_id, + operation_request_id=body.operation_request_id, + expected_repo_state_digest=body.expected_repo_state_digest, + paths=tuple(root / path for path in body.paths), + path_sources=body.path_sources, + target_role=body.target_role, + target_id=body.target_id, + actor_id=body.actor_id, + trigger=body.trigger, + message=body.message, + ) + except Exception as exc: + raise _git_error(exc) from exc + return { + "checkpoint_id": checkpoint.checkpoint_id, + "repository_id": checkpoint.repository_id, + "commit_oid": checkpoint.commit_oid, + "parent_oid": checkpoint.parent_oid, + "paths": list(checkpoint.paths), + "created_at": checkpoint.created_at, + } + + +@router.post("/spaces/{space_id}/git/restore", status_code=201) +async def git_restore_candidate( + space_id: str, + body: GitRestoreBody, +): + bound_root = _binding_root( + space_id=space_id, + email=body.email, + user_id=body.user_id, + ) + service = _service() + checkpoint = service.journal.get_git_checkpoint(body.checkpoint_id) + repository = service.journal.get_space_git_repository(space_id=space_id) + if ( + repository is None + or checkpoint is None + or checkpoint.repository_id != repository.repository_id + ): + raise HTTPException(status_code=404, detail="Checkpoint not found") + _assert_repository_binding(repository, bound_root) + try: + candidate = service.prepare_restore_candidate( + body.checkpoint_id, + operation_request_id=body.operation_request_id, + expected_repo_state_digest=body.expected_repo_state_digest, + ) + except Exception as exc: + raise _git_error(exc) from exc + return { + "checkpoint_id": body.checkpoint_id, + "repository_id": repository.repository_id, + "candidate_ref": candidate.ref_name, + "commit_oid": candidate.commit_oid, + "applied_to_user_worktree": False, + } diff --git a/backend/app/router.py b/backend/app/router.py index 29267e05a..2d6b2e47e 100644 --- a/backend/app/router.py +++ b/backend/app/router.py @@ -36,6 +36,7 @@ from app.controller import ( task_controller, tool_controller, workspace_controller, + workspace_git_controller, ) logger = logging.getLogger("router") @@ -122,6 +123,12 @@ def register_routers(app: FastAPI, prefix: str = "") -> None: "tags": ["workspace"], "description": "Space-level local workspace binding", }, + { + "router": workspace_git_controller.router, + "tags": ["workspace-git"], + "description": "Authenticated local Space Git operations", + "self_authenticated": True, + }, ] app.include_router(health_controller.router, tags=["Health"]) diff --git a/backend/app/run_journal/__init__.py b/backend/app/run_journal/__init__.py index 9d6d13a6f..f23c5a6cb 100644 --- a/backend/app/run_journal/__init__.py +++ b/backend/app/run_journal/__init__.py @@ -21,6 +21,9 @@ from app.run_journal.models import ( CommandResultSyncBatch, CommittedRunEvent, EffectiveEnvironmentSpecRecord, + GitCheckpointRecord, + GitOperationRecord, + GitRepositoryRecord, RemoteCommandInboxRecord, RunAttemptRecord, RunEventDraft, @@ -64,6 +67,9 @@ __all__ = [ "CommandResultSyncBatch", "EventRecorder", "EffectiveEnvironmentSpecRecord", + "GitCheckpointRecord", + "GitOperationRecord", + "GitRepositoryRecord", "IdempotencyConflictError", "InvalidRunTransitionError", "OptimisticConcurrencyError", diff --git a/backend/app/run_journal/models.py b/backend/app/run_journal/models.py index 83981b147..ff29f2074 100644 --- a/backend/app/run_journal/models.py +++ b/backend/app/run_journal/models.py @@ -143,6 +143,56 @@ class EffectiveEnvironmentSpecRecord: created_at: float +@dataclass(frozen=True) +class GitRepositoryRecord: + repository_id: str + space_id: str + repository_role: str + root_path: str + root_path_digest: str + ownership: str + state: str + version_coverage: str + hooks_mode: str + repo_subdir: str | None + version: int + created_at: float + updated_at: float + + +@dataclass(frozen=True) +class GitOperationRecord: + operation_id: str + repository_id: str + request_id: str + operation_type: str + payload_digest: str + status: str + expected_repo_state_digest: str | None + observed_repo_state_digest: str | None + result: dict[str, Any] | None + error_code: str | None + error_message: str | None + created_at: float + updated_at: float + + +@dataclass(frozen=True) +class GitCheckpointRecord: + checkpoint_id: str + repository_id: str + operation_id: str + target_role: str + target_id: str + commit_oid: str + parent_oid: str | None + paths: tuple[str, ...] + actor_id: str + trigger: str + message: str + created_at: float + + @dataclass(frozen=True) class ToolCallRecord: tool_call_id: str diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index 989fb4fc2..e28263c11 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -42,6 +42,9 @@ from app.run_journal.models import ( CommandResultSyncBatch, CommittedRunEvent, EffectiveEnvironmentSpecRecord, + GitCheckpointRecord, + GitOperationRecord, + GitRepositoryRecord, RemoteCommandInboxRecord, RunAttemptRecord, RunEventDraft, @@ -76,7 +79,7 @@ from app.workspace_config.models import ( canonical_json, ) -SCHEMA_VERSION = 6 +SCHEMA_VERSION = 7 logger = logging.getLogger("run_journal") _MIGRATION_V1 = """ @@ -431,6 +434,113 @@ PRAGMA user_version = 6; COMMIT; """ +_MIGRATION_V7 = """ +BEGIN IMMEDIATE; + +CREATE TABLE git_repositories ( + repository_id TEXT PRIMARY KEY, + space_id TEXT NOT NULL, + repository_role TEXT NOT NULL CHECK ( + repository_role IN ('content', 'configuration') + ), + root_path TEXT NOT NULL, + root_path_digest TEXT NOT NULL CHECK (length(root_path_digest) = 64), + ownership TEXT NOT NULL CHECK ( + ownership IN ('eigent_owned', 'adopted') + ), + state TEXT NOT NULL CHECK ( + state IN ('ready', 'not_enabled', 'needs_attention', 'degraded') + ), + version_coverage TEXT NOT NULL CHECK ( + version_coverage IN ('full', 'managed_files_only', 'degraded') + ), + hooks_mode TEXT NOT NULL DEFAULT 'disabled' CHECK ( + hooks_mode IN ('disabled', 'trusted') + ), + repo_subdir TEXT, + version INTEGER NOT NULL DEFAULT 0 CHECK (version >= 0), + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(space_id, repository_role) +); + +CREATE TABLE git_operations ( + operation_id TEXT PRIMARY KEY, + repository_id TEXT NOT NULL REFERENCES git_repositories( + repository_id + ) ON DELETE RESTRICT, + request_id TEXT NOT NULL, + operation_type TEXT NOT NULL, + payload_digest TEXT NOT NULL CHECK (length(payload_digest) = 64), + status TEXT NOT NULL CHECK ( + status IN ( + 'prepared', 'dispatched', 'completed', 'failed', + 'outcome_unknown' + ) + ), + expected_repo_state_digest TEXT, + observed_repo_state_digest TEXT, + result_json TEXT, + error_code TEXT, + error_message TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + UNIQUE(repository_id, request_id) +); + +CREATE INDEX git_operations_reconcile_idx +ON git_operations(status, updated_at, repository_id); + +CREATE TABLE git_checkpoints ( + checkpoint_id TEXT PRIMARY KEY, + repository_id TEXT NOT NULL REFERENCES git_repositories( + repository_id + ) ON DELETE RESTRICT, + operation_id TEXT NOT NULL UNIQUE REFERENCES git_operations( + operation_id + ) ON DELETE RESTRICT, + target_role TEXT NOT NULL CHECK ( + target_role IN ('user', 'project', 'run', 'agent') + ), + target_id TEXT NOT NULL, + commit_oid TEXT NOT NULL, + parent_oid TEXT, + paths_json TEXT NOT NULL, + actor_id TEXT NOT NULL, + trigger TEXT NOT NULL, + message TEXT NOT NULL, + created_at REAL NOT NULL +); + +CREATE INDEX git_checkpoints_repository_created_idx +ON git_checkpoints(repository_id, created_at DESC); + +CREATE TABLE git_managed_paths ( + repository_id TEXT NOT NULL REFERENCES git_repositories( + repository_id + ) ON DELETE CASCADE, + relative_path TEXT NOT NULL, + source TEXT NOT NULL CHECK ( + source IN ( + 'agent_created', 'agent_modified', 'user_selected', + 'configuration', 'overlay_preimage' + ) + ), + first_checkpoint_id TEXT REFERENCES git_checkpoints( + checkpoint_id + ) ON DELETE SET NULL, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + PRIMARY KEY(repository_id, relative_path) +); + +INSERT OR IGNORE INTO run_journal_migrations(version, applied_at) +VALUES (7, CAST(strftime('%s', 'now') AS REAL)); + +PRAGMA user_version = 7; +COMMIT; +""" + class RunJournalError(RuntimeError): """Base error for local RunJournal operations.""" @@ -1006,6 +1116,609 @@ class SQLiteRunJournal: else None ) + def put_git_repository( + self, + *, + repository_id: str, + space_id: str, + repository_role: str, + root_path: str, + root_path_digest: str, + ownership: str, + state: str, + version_coverage: str, + hooks_mode: str = "disabled", + repo_subdir: str | None = None, + now: float | None = None, + ) -> GitRepositoryRecord: + timestamp = now if now is not None else time.time() + immutable_expected = ( + repository_id, + space_id, + repository_role, + root_path, + root_path_digest, + ownership, + version_coverage, + hooks_mode, + repo_subdir, + ) + with self._write_transaction() as connection: + by_identity = connection.execute( + "SELECT * FROM git_repositories WHERE repository_id = ?", + (repository_id,), + ).fetchone() + by_role = connection.execute( + """ + SELECT * FROM git_repositories + WHERE space_id = ? AND repository_role = ? + """, + (space_id, repository_role), + ).fetchone() + row = by_identity or by_role + if row is not None: + actual = ( + row["repository_id"], + row["space_id"], + row["repository_role"], + row["root_path"], + row["root_path_digest"], + row["ownership"], + row["version_coverage"], + row["hooks_mode"], + row["repo_subdir"], + ) + if actual != immutable_expected: + raise IdempotencyConflictError( + f"Git repository ownership for Space {space_id!r} " + "conflicts with the persisted binding" + ) + if row["state"] != state: + connection.execute( + """ + UPDATE git_repositories + SET state = ?, version = version + 1, updated_at = ? + WHERE repository_id = ? + """, + (state, timestamp, row["repository_id"]), + ) + row = connection.execute( + """ + SELECT * FROM git_repositories + WHERE repository_id = ? + """, + (row["repository_id"],), + ).fetchone() + assert row is not None + return self._git_repository_from_row(row) + connection.execute( + """ + INSERT INTO git_repositories( + repository_id, space_id, repository_role, root_path, + root_path_digest, ownership, state, version_coverage, + hooks_mode, repo_subdir, version, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?) + """, + ( + *immutable_expected[:6], + state, + *immutable_expected[6:], + timestamp, + timestamp, + ), + ) + row = connection.execute( + "SELECT * FROM git_repositories WHERE repository_id = ?", + (repository_id,), + ).fetchone() + assert row is not None + return self._git_repository_from_row(row) + + def update_git_repository_state( + self, + repository_id: str, + *, + state: str, + expected_version: int, + now: float | None = None, + ) -> GitRepositoryRecord: + if state not in { + "ready", + "not_enabled", + "needs_attention", + "degraded", + }: + raise ValueError(f"unsupported Git repository state {state!r}") + timestamp = now if now is not None else time.time() + with self._write_transaction() as connection: + row = connection.execute( + "SELECT * FROM git_repositories WHERE repository_id = ?", + (repository_id,), + ).fetchone() + if row is None: + raise ValueError(f"unknown Git repository {repository_id!r}") + if row["state"] == state: + return self._git_repository_from_row(row) + if int(row["version"]) != expected_version: + raise IdempotencyConflictError( + f"Git repository {repository_id!r} changed concurrently" + ) + connection.execute( + """ + UPDATE git_repositories + SET state = ?, version = version + 1, updated_at = ? + WHERE repository_id = ? AND version = ? + """, + (state, timestamp, repository_id, expected_version), + ) + row = connection.execute( + "SELECT * FROM git_repositories WHERE repository_id = ?", + (repository_id,), + ).fetchone() + assert row is not None + return self._git_repository_from_row(row) + + def get_git_repository( + self, repository_id: str + ) -> GitRepositoryRecord | None: + with self._lock: + row = self._connection.execute( + "SELECT * FROM git_repositories WHERE repository_id = ?", + (repository_id,), + ).fetchone() + return ( + self._git_repository_from_row(row) if row is not None else None + ) + + def get_space_git_repository( + self, + *, + space_id: str, + repository_role: str = "content", + ) -> GitRepositoryRecord | None: + with self._lock: + row = self._connection.execute( + """ + SELECT * FROM git_repositories + WHERE space_id = ? AND repository_role = ? + """, + (space_id, repository_role), + ).fetchone() + return ( + self._git_repository_from_row(row) if row is not None else None + ) + + def begin_git_operation( + self, + *, + operation_id: str, + repository_id: str, + request_id: str, + operation_type: str, + payload_digest: str, + expected_repo_state_digest: str | None, + now: float | None = None, + ) -> GitOperationRecord: + if len(payload_digest) != 64: + raise ValueError("Git operation payload digest must be SHA-256") + timestamp = now if now is not None else time.time() + expected = ( + operation_id, + repository_id, + request_id, + operation_type, + payload_digest, + expected_repo_state_digest, + ) + with self._write_transaction() as connection: + if ( + connection.execute( + "SELECT 1 FROM git_repositories WHERE repository_id = ?", + (repository_id,), + ).fetchone() + is None + ): + raise ValueError(f"unknown Git repository {repository_id!r}") + row = connection.execute( + """ + SELECT * FROM git_operations + WHERE operation_id = ? OR ( + repository_id = ? AND request_id = ? + ) + """, + (operation_id, repository_id, request_id), + ).fetchone() + if row is not None: + actual = ( + row["operation_id"], + row["repository_id"], + row["request_id"], + row["operation_type"], + row["payload_digest"], + row["expected_repo_state_digest"], + ) + if actual != expected: + raise IdempotencyConflictError( + f"Git operation request {request_id!r} was reused " + "with a different action" + ) + return self._git_operation_from_row(row) + connection.execute( + """ + INSERT INTO git_operations( + operation_id, repository_id, request_id, + operation_type, payload_digest, status, + expected_repo_state_digest, observed_repo_state_digest, + result_json, error_code, error_message, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, 'prepared', ?, NULL, NULL, NULL, + NULL, ?, ?) + """, + (*expected, timestamp, timestamp), + ) + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + assert row is not None + return self._git_operation_from_row(row) + + def mark_git_operation_dispatched( + self, + operation_id: str, + *, + observed_repo_state_digest: str, + now: float | None = None, + ) -> GitOperationRecord: + timestamp = now if now is not None else time.time() + with self._write_transaction() as connection: + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if row is None: + raise ValueError(f"unknown Git operation {operation_id!r}") + if row["status"] in {"dispatched", "completed"}: + return self._git_operation_from_row(row) + if row["status"] != "prepared": + raise InvalidRunTransitionError( + f"Git operation {operation_id!r} cannot dispatch from " + f"{row['status']!r}" + ) + connection.execute( + """ + UPDATE git_operations + SET status = 'dispatched', observed_repo_state_digest = ?, + updated_at = ? + WHERE operation_id = ? AND status = 'prepared' + """, + (observed_repo_state_digest, timestamp, operation_id), + ) + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + assert row is not None + return self._git_operation_from_row(row) + + def complete_git_operation( + self, + operation_id: str, + *, + result: dict[str, Any], + observed_repo_state_digest: str, + now: float | None = None, + ) -> GitOperationRecord: + timestamp = now if now is not None else time.time() + result_json = canonical_json(result) + with self._write_transaction() as connection: + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if row is None: + raise ValueError(f"unknown Git operation {operation_id!r}") + if row["status"] == "completed": + if row["result_json"] != result_json: + raise IdempotencyConflictError( + f"Git operation {operation_id!r} completed with a " + "different result" + ) + return self._git_operation_from_row(row) + if row["status"] != "dispatched": + raise InvalidRunTransitionError( + f"Git operation {operation_id!r} cannot complete from " + f"{row['status']!r}" + ) + connection.execute( + """ + UPDATE git_operations + SET status = 'completed', result_json = ?, + observed_repo_state_digest = ?, error_code = NULL, + error_message = NULL, updated_at = ? + WHERE operation_id = ? + """, + ( + result_json, + observed_repo_state_digest, + timestamp, + operation_id, + ), + ) + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + assert row is not None + return self._git_operation_from_row(row) + + def fail_git_operation( + self, + operation_id: str, + *, + error_code: str, + error_message: str, + outcome_unknown: bool = False, + now: float | None = None, + ) -> GitOperationRecord: + timestamp = now if now is not None else time.time() + target = "outcome_unknown" if outcome_unknown else "failed" + with self._write_transaction() as connection: + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if row is None: + raise ValueError(f"unknown Git operation {operation_id!r}") + if row["status"] == "completed": + return self._git_operation_from_row(row) + if row["status"] in {"failed", "outcome_unknown"}: + if row["status"] == target: + return self._git_operation_from_row(row) + raise InvalidRunTransitionError( + f"Git operation {operation_id!r} cannot transition from " + f"{row['status']!r} to {target!r}" + ) + if row["status"] == "prepared" and outcome_unknown: + raise InvalidRunTransitionError( + "a Git operation cannot become outcome_unknown before " + "dispatch" + ) + connection.execute( + """ + UPDATE git_operations + SET status = ?, error_code = ?, error_message = ?, + updated_at = ? + WHERE operation_id = ? + """, + ( + target, + error_code, + error_message, + timestamp, + operation_id, + ), + ) + row = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + assert row is not None + return self._git_operation_from_row(row) + + def get_git_operation( + self, operation_id: str + ) -> GitOperationRecord | None: + with self._lock: + row = self._connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + return ( + self._git_operation_from_row(row) if row is not None else None + ) + + def complete_git_checkpoint( + self, + *, + checkpoint_id: str, + operation_id: str, + repository_id: str, + target_role: str, + target_id: str, + commit_oid: str, + parent_oid: str | None, + paths: tuple[str, ...], + managed_path_sources: dict[str, str], + actor_id: str, + trigger: str, + message: str, + observed_repo_state_digest: str, + now: float | None = None, + ) -> GitCheckpointRecord: + if not paths or tuple(sorted(set(paths))) != paths: + raise ValueError( + "checkpoint paths must be non-empty, unique, and sorted" + ) + if set(managed_path_sources) != set(paths): + raise ValueError( + "managed path sources must match checkpoint paths" + ) + timestamp = now if now is not None else time.time() + paths_json = canonical_json(list(paths)) + result = { + "checkpoint_id": checkpoint_id, + "commit_oid": commit_oid, + "parent_oid": parent_oid, + "paths": list(paths), + } + result_json = canonical_json(result) + with self._write_transaction() as connection: + operation = connection.execute( + "SELECT * FROM git_operations WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if ( + operation is None + or operation["repository_id"] != repository_id + ): + raise ValueError("checkpoint operation/repository mismatch") + existing = connection.execute( + """ + SELECT * FROM git_checkpoints + WHERE checkpoint_id = ? OR operation_id = ? + """, + (checkpoint_id, operation_id), + ).fetchone() + if existing is not None: + expected = ( + checkpoint_id, + repository_id, + operation_id, + target_role, + target_id, + commit_oid, + parent_oid, + paths_json, + actor_id, + trigger, + message, + ) + actual = tuple( + existing[column] + for column in ( + "checkpoint_id", + "repository_id", + "operation_id", + "target_role", + "target_id", + "commit_oid", + "parent_oid", + "paths_json", + "actor_id", + "trigger", + "message", + ) + ) + if actual != expected: + raise IdempotencyConflictError( + f"checkpoint {checkpoint_id!r} conflicts with its " + "persisted result" + ) + return self._git_checkpoint_from_row(existing) + if operation["status"] != "dispatched": + raise InvalidRunTransitionError( + f"Git operation {operation_id!r} cannot create a " + f"checkpoint from {operation['status']!r}" + ) + connection.execute( + """ + INSERT INTO git_checkpoints( + checkpoint_id, repository_id, operation_id, target_role, + target_id, commit_oid, parent_oid, paths_json, actor_id, + trigger, message, created_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + checkpoint_id, + repository_id, + operation_id, + target_role, + target_id, + commit_oid, + parent_oid, + paths_json, + actor_id, + trigger, + message, + timestamp, + ), + ) + for relative_path, source in managed_path_sources.items(): + connection.execute( + """ + INSERT INTO git_managed_paths( + repository_id, relative_path, source, + first_checkpoint_id, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(repository_id, relative_path) DO UPDATE SET + updated_at = excluded.updated_at + """, + ( + repository_id, + relative_path, + source, + checkpoint_id, + timestamp, + timestamp, + ), + ) + connection.execute( + """ + UPDATE git_operations + SET status = 'completed', result_json = ?, + observed_repo_state_digest = ?, error_code = NULL, + error_message = NULL, updated_at = ? + WHERE operation_id = ? + """, + ( + result_json, + observed_repo_state_digest, + timestamp, + operation_id, + ), + ) + row = connection.execute( + "SELECT * FROM git_checkpoints WHERE checkpoint_id = ?", + (checkpoint_id,), + ).fetchone() + assert row is not None + return self._git_checkpoint_from_row(row) + + def get_git_checkpoint( + self, checkpoint_id: str + ) -> GitCheckpointRecord | None: + with self._lock: + row = self._connection.execute( + "SELECT * FROM git_checkpoints WHERE checkpoint_id = ?", + (checkpoint_id,), + ).fetchone() + return ( + self._git_checkpoint_from_row(row) if row is not None else None + ) + + def list_git_checkpoints( + self, + repository_id: str, + *, + limit: int = 100, + ) -> list[GitCheckpointRecord]: + if limit < 1: + raise ValueError("checkpoint query limit must be positive") + with self._lock: + rows = self._connection.execute( + """ + SELECT * FROM git_checkpoints + WHERE repository_id = ? + ORDER BY created_at DESC, checkpoint_id DESC + LIMIT ? + """, + (repository_id, limit), + ).fetchall() + return [self._git_checkpoint_from_row(row) for row in rows] + + def list_git_managed_paths(self, repository_id: str) -> tuple[str, ...]: + with self._lock: + rows = self._connection.execute( + """ + SELECT relative_path FROM git_managed_paths + WHERE repository_id = ? + ORDER BY relative_path + """, + (repository_id,), + ).fetchall() + return tuple(row["relative_path"] for row in rows) + def ensure_run( self, *, @@ -3825,6 +4538,8 @@ class SQLiteRunJournal: self._connection.executescript(_MIGRATION_V5) if version < 6: self._connection.executescript(_MIGRATION_V6) + if version < 7: + self._connection.executescript(_MIGRATION_V7) @contextmanager def _write_transaction(self) -> Iterator[sqlite3.Connection]: @@ -4135,6 +4850,63 @@ class SQLiteRunJournal: created_at=float(row["created_at"]), ) + @staticmethod + def _git_repository_from_row(row: sqlite3.Row) -> GitRepositoryRecord: + return GitRepositoryRecord( + repository_id=row["repository_id"], + space_id=row["space_id"], + repository_role=row["repository_role"], + root_path=row["root_path"], + root_path_digest=row["root_path_digest"], + ownership=row["ownership"], + state=row["state"], + version_coverage=row["version_coverage"], + hooks_mode=row["hooks_mode"], + repo_subdir=row["repo_subdir"], + version=int(row["version"]), + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + ) + + @staticmethod + def _git_operation_from_row(row: sqlite3.Row) -> GitOperationRecord: + return GitOperationRecord( + operation_id=row["operation_id"], + repository_id=row["repository_id"], + request_id=row["request_id"], + operation_type=row["operation_type"], + payload_digest=row["payload_digest"], + status=row["status"], + expected_repo_state_digest=row["expected_repo_state_digest"], + observed_repo_state_digest=row["observed_repo_state_digest"], + result=( + json.loads(row["result_json"]) + if row["result_json"] is not None + else None + ), + error_code=row["error_code"], + error_message=row["error_message"], + created_at=float(row["created_at"]), + updated_at=float(row["updated_at"]), + ) + + @staticmethod + def _git_checkpoint_from_row(row: sqlite3.Row) -> GitCheckpointRecord: + return GitCheckpointRecord( + checkpoint_id=row["checkpoint_id"], + repository_id=row["repository_id"], + operation_id=row["operation_id"], + target_role=row["target_role"], + target_id=row["target_id"], + commit_oid=row["commit_oid"], + parent_oid=row["parent_oid"], + paths=tuple(json.loads(row["paths_json"])), + actor_id=row["actor_id"], + trigger=row["trigger"], + message=row["message"], + created_at=float(row["created_at"]), + ) + @staticmethod def _tool_call_from_row(row: sqlite3.Row) -> ToolCallRecord: return ToolCallRecord( diff --git a/backend/app/workspace_git/__init__.py b/backend/app/workspace_git/__init__.py index bf38c8bbd..599c1fdb2 100644 --- a/backend/app/workspace_git/__init__.py +++ b/backend/app/workspace_git/__init__.py @@ -18,22 +18,46 @@ from app.workspace_git.backend import ( GitCommandError, GitCommandResult, NestedRepositoryError, + RepositoryDiagnostics, RepositoryProbe, + RepoStateToken, ) from app.workspace_git.configuration import ( ConfigurationRepositoryError, ConfigurationRepositoryResult, ConfigurationRepositoryService, ) +from app.workspace_git.content import ( + ContentRepositoryConsentRequired, + ContentRepositoryError, + ContentRepositoryInspection, + ContentRepositoryResult, + ContentRepositoryService, + ContentRepositoryStatus, + NoCheckpointChangesError, + RepositoryStateChangedError, + RestoreCandidate, +) __all__ = [ "ConfigurationRepositoryError", "ConfigurationRepositoryResult", "ConfigurationRepositoryService", + "ContentRepositoryConsentRequired", + "ContentRepositoryError", + "ContentRepositoryInspection", + "ContentRepositoryResult", + "ContentRepositoryService", + "ContentRepositoryStatus", "GitBackend", "GitBackendError", "GitCommandError", "GitCommandResult", "NestedRepositoryError", + "NoCheckpointChangesError", + "RepoStateToken", + "RepositoryDiagnostics", + "RepositoryStateChangedError", "RepositoryProbe", + "RestoreCandidate", ] diff --git a/backend/app/workspace_git/backend.py b/backend/app/workspace_git/backend.py index c63077abb..29a8c4ed7 100644 --- a/backend/app/workspace_git/backend.py +++ b/backend/app/workspace_git/backend.py @@ -16,7 +16,9 @@ from __future__ import annotations +import hashlib import os +import re import shutil import subprocess from dataclasses import dataclass @@ -76,6 +78,35 @@ class RepositoryProbe: branch: str | None +@dataclass(frozen=True) +class RepoStateToken: + head_oid: str | None + branch_or_detached_head: str + index_digest: str + operation_state: str + + @property + def digest(self) -> str: + payload = "\0".join( + ( + self.head_oid or "unborn", + self.branch_or_detached_head, + self.index_digest, + self.operation_state, + ) + ) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +@dataclass(frozen=True) +class RepositoryDiagnostics: + healthy: bool + issues: tuple[str, ...] + state_token: RepoStateToken + has_submodules: bool + has_remotes: bool + + class GitBackend: """Small typed backend; arbitrary command execution is intentionally absent.""" @@ -180,6 +211,56 @@ class GitBackend: ) return result.stdout.strip() if result.returncode == 0 else None + def repo_state_token(self, repository_root: Path) -> RepoStateToken: + root = repository_root.expanduser().resolve() + probe = self.probe(root) + if not probe.is_repository or not probe.owns_requested_root: + raise GitBackendError(f"not an owned Git root: {root}") + index = self._run(root, ("ls-files", "--stage", "-z")) + status = self._run( + root, + ("status", "--porcelain=v1", "-z", "--untracked-files=all"), + ) + worktree_metadata = self._status_metadata(root, status.stdout) + index_digest = hashlib.sha256( + ( + index.stdout + "\0" + status.stdout + "\0" + worktree_metadata + ).encode("utf-8") + ).hexdigest() + return RepoStateToken( + head_oid=probe.head_oid, + branch_or_detached_head=(probe.branch or "DETACHED"), + index_digest=index_digest, + operation_state=self._operation_state(root), + ) + + def diagnostics(self, repository_root: Path) -> RepositoryDiagnostics: + root = repository_root.expanduser().resolve() + token = self.repo_state_token(root) + issues: list[str] = [] + if token.operation_state != "clean": + issues.append(f"operation_in_progress:{token.operation_state}") + if token.head_oid is not None: + connectivity = self._run( + root, + ("cat-file", "-e", f"{token.head_oid}^{{commit}}"), + check=False, + ) + if connectivity.returncode != 0: + issues.append("object_database_unhealthy") + staged = self._run(root, ("ls-files", "--stage")) + has_submodules = any( + line.startswith("160000 ") for line in staged.stdout.splitlines() + ) + remotes = self._run(root, ("remote",)) + return RepositoryDiagnostics( + healthy=not issues, + issues=tuple(issues), + state_token=token, + has_submodules=has_submodules, + has_remotes=bool(remotes.stdout.strip()), + ) + def changed_paths( self, repository_root: Path, @@ -224,6 +305,7 @@ class GitBackend: if not message.strip(): raise ValueError("Git commit message is required") pathspecs = self._relative_pathspecs(repository_root, paths) + self._assert_no_clean_filters(repository_root, pathspecs) self._run(repository_root, ("add", "--", *pathspecs)) staged = self._run( repository_root, @@ -278,6 +360,125 @@ class GitBackend: ) return tuple(line for line in result.stdout.splitlines() if line) + def relative_paths( + self, + repository_root: Path, + paths: tuple[Path, ...], + ) -> tuple[str, ...]: + return self._relative_pathspecs(repository_root, paths) + + def diff_paths( + self, + repository_root: Path, + paths: tuple[Path, ...], + *, + cached: bool = False, + source: str | None = None, + ) -> str: + pathspecs = self._relative_pathspecs(repository_root, paths) + args = ["diff", "--no-ext-diff", "--no-textconv"] + if cached: + args.append("--cached") + if source is not None: + self._validate_object_name(source) + args.append(source) + args.extend(("--", *pathspecs)) + return self._run(repository_root, tuple(args)).stdout + + def commit_parent( + self, + repository_root: Path, + commit_oid: str, + ) -> str | None: + self._validate_object_name(commit_oid) + result = self._run( + repository_root, + ("rev-parse", f"{commit_oid}^"), + check=False, + ) + return result.stdout.strip() if result.returncode == 0 else None + + def find_commit_by_operation( + self, + repository_root: Path, + operation_id: str, + ) -> str | None: + if not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", operation_id): + raise ValueError("invalid Git operation id") + result = self._run( + repository_root, + ( + "log", + "--all", + "--fixed-strings", + f"--grep=Eigent-Operation: {operation_id}", + "-1", + "--format=%H", + ), + check=False, + ) + value = result.stdout.strip() + return value if result.returncode == 0 and value else None + + def update_eigent_ref( + self, + repository_root: Path, + ref_name: str, + commit_oid: str, + *, + expected_oid: str | None = None, + ) -> str: + if not ref_name.startswith("refs/eigent/") or not re.fullmatch( + r"refs/eigent/[A-Za-z0-9._/-]+", ref_name + ): + raise ValueError("ref must be inside refs/eigent/") + self._validate_object_name(commit_oid) + args = ["update-ref", ref_name, commit_oid] + if expected_oid is not None: + self._validate_object_name(expected_oid) + args.append(expected_oid) + self._run(repository_root, tuple(args)) + return commit_oid + + def ref_oid( + self, + repository_root: Path, + ref_name: str, + ) -> str | None: + if not ref_name.startswith("refs/eigent/"): + raise ValueError("only Eigent-owned refs may be queried") + result = self._run( + repository_root, + ("rev-parse", "--verify", ref_name), + check=False, + ) + return result.stdout.strip() if result.returncode == 0 else None + + def _operation_state(self, repository_root: Path) -> str: + markers = ( + ("MERGE_HEAD", "merge"), + ("rebase-merge", "rebase"), + ("rebase-apply", "rebase"), + ("CHERRY_PICK_HEAD", "cherry-pick"), + ("REVERT_HEAD", "revert"), + ) + for marker, state in markers: + result = self._run( + repository_root, + ("rev-parse", "--git-path", marker), + ) + path = Path(result.stdout.strip()) + if not path.is_absolute(): + path = repository_root / path + if path.exists(): + return state + return "clean" + + @staticmethod + def _validate_object_name(value: str) -> None: + if not re.fullmatch(r"[0-9a-fA-F]{4,64}", value): + raise ValueError("Git object id must be hexadecimal") + def _relative_pathspecs( self, repository_root: Path, @@ -288,7 +489,10 @@ class GitBackend: root = repository_root.expanduser().resolve() pathspecs: list[str] = [] for path in paths: - resolved = path.expanduser().resolve() + candidate = path.expanduser() + if not candidate.is_absolute(): + candidate = root / candidate + resolved = candidate.resolve() try: relative = resolved.relative_to(root) except ValueError as exc: @@ -302,6 +506,55 @@ class GitBackend: pathspecs.append(relative.as_posix()) return tuple(pathspecs) + @staticmethod + def _status_metadata(repository_root: Path, status_output: str) -> str: + records = status_output.split("\0") + metadata: list[str] = [] + skip_next = False + for record in records: + if not record: + continue + if skip_next: + skip_next = False + continue + if len(record) < 4: + continue + state = record[:2] + relative_path = record[3:] + if "R" in state or "C" in state: + skip_next = True + path = repository_root / relative_path + try: + stat = path.lstat() + value = ( + f"{relative_path}\0{stat.st_mode}\0{stat.st_size}\0" + f"{stat.st_mtime_ns}\0{stat.st_ino}" + ) + except FileNotFoundError: + value = f"{relative_path}\0missing" + metadata.append(value) + return "\0".join(sorted(metadata)) + + def _assert_no_clean_filters( + self, + repository_root: Path, + pathspecs: tuple[str, ...], + ) -> None: + attributes = self._run( + repository_root, + ("check-attr", "-a", "-z", "--", *pathspecs), + ).stdout.split("\0") + for index in range(0, len(attributes) - 2, 3): + path, attribute, value = attributes[index : index + 3] + if attribute == "filter" and value not in { + "unspecified", + "unset", + "", + }: + raise GitBackendError( + f"refusing to execute clean filter {value!r} for {path!r}" + ) + def _run( self, cwd: Path, @@ -343,6 +596,10 @@ class GitBackend: self.git_executable, "-c", f"core.hooksPath={self.hooks_path}", + "-c", + "core.fsmonitor=false", + "-c", + "core.untrackedCache=false", "-C", str(cwd), *args, diff --git a/backend/app/workspace_git/content.py b/backend/app/workspace_git/content.py new file mode 100644 index 000000000..5c3ee3bde --- /dev/null +++ b/backend/app/workspace_git/content.py @@ -0,0 +1,679 @@ +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. ========= + +"""Content Repository ownership, checkpoints, and safe restore candidates.""" + +from __future__ import annotations + +import os +import re +import threading +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path + +from app.run_journal import ( + GitCheckpointRecord, + GitOperationRecord, + GitRepositoryRecord, + SQLiteRunJournal, +) +from app.workspace_config import canonical_digest +from app.workspace_git.backend import ( + GitBackend, + GitBackendError, + NestedRepositoryError, + RepositoryDiagnostics, + RepositoryProbe, +) + +try: + import fcntl +except ImportError: # pragma: no cover - Windows fallback + fcntl = None + + +class ContentRepositoryError(RuntimeError): + """Base error for Content Repository operations.""" + + +class ContentRepositoryConsentRequired(ContentRepositoryError): + """Raised before initializing a user-owned folder without consent.""" + + +class RepositoryStateChangedError(ContentRepositoryError): + """Raised when optimistic RepoStateToken validation fails.""" + + +class NoCheckpointChangesError(ContentRepositoryError): + """Raised when none of the explicit checkpoint paths changed.""" + + +@dataclass(frozen=True) +class ContentRepositoryInspection: + probe: RepositoryProbe + diagnostics: RepositoryDiagnostics | None + enablement: str + consent_required: bool + + +@dataclass(frozen=True) +class ContentRepositoryResult: + repository: GitRepositoryRecord + probe: RepositoryProbe + diagnostics: RepositoryDiagnostics + initialized: bool + + +@dataclass(frozen=True) +class ContentRepositoryStatus: + repository: GitRepositoryRecord + diagnostics: RepositoryDiagnostics + managed_paths: tuple[str, ...] + + +@dataclass(frozen=True) +class RestoreCandidate: + operation: GitOperationRecord + checkpoint: GitCheckpointRecord + ref_name: str + commit_oid: str + + +_SAFE_IDENTIFIER = re.compile(r"[A-Za-z0-9_.:-]{1,128}") +_MANAGED_SOURCES = { + "agent_created", + "agent_modified", + "user_selected", + "configuration", + "overlay_preimage", +} +_TARGET_ROLES = {"user", "project", "run", "agent"} + + +class ContentRepositoryService: + """Single typed owner for local Content Repository mutations. + + Checkpoints only stage caller-supplied paths. Restore creates an + Eigent-owned recovery ref; it never checks out, resets, or cleans the User + Worktree. + """ + + def __init__( + self, + journal: SQLiteRunJournal, + *, + state_root: Path, + git_backend: GitBackend | None = None, + ) -> None: + self.journal = journal + self.state_root = state_root.expanduser().resolve() + self.git = git_backend or GitBackend() + self._lock = threading.RLock() + + def inspect(self, space_root: Path) -> ContentRepositoryInspection: + root = space_root.expanduser().resolve() + if not root.is_dir(): + raise ContentRepositoryError( + f"Space root is not a directory: {space_root}" + ) + probe = self.git.probe(root) + if probe.nested_in_parent: + return ContentRepositoryInspection( + probe=probe, + diagnostics=None, + enablement="nested_repository_requires_binding", + consent_required=True, + ) + if not probe.is_repository: + return ContentRepositoryInspection( + probe=probe, + diagnostics=None, + enablement="not_enabled", + consent_required=True, + ) + diagnostics = self.git.diagnostics(root) + return ContentRepositoryInspection( + probe=probe, + diagnostics=diagnostics, + enablement=("ready" if diagnostics.healthy else "needs_attention"), + consent_required=False, + ) + + def bootstrap( + self, + *, + space_id: str, + space_root: Path, + allow_init: bool, + eigent_owned_space: bool = False, + repo_subdir: str | None = None, + ) -> ContentRepositoryResult: + self._validate_identifier("space_id", space_id) + root = space_root.expanduser().resolve() + if not root.is_dir(): + raise ContentRepositoryError( + f"Space root is not a directory: {space_root}" + ) + lock_path = self._lock_path(space_id) + with self._repository_lock(lock_path): + before = self.git.probe(root) + if before.nested_in_parent: + raise NestedRepositoryError( + "Space root is inside a parent repository; explicit " + "parent binding and repo_subdir confirmation are required" + ) + initialized = False + if not before.is_repository: + if not allow_init: + raise ContentRepositoryConsentRequired( + "enabling local version management requires explicit " + "consent for this folder" + ) + self.git.init_repository(root) + initialized = True + probe = self.git.probe(root) + if not probe.is_repository or not probe.owns_requested_root: + raise ContentRepositoryError( + "Content Repository did not resolve to the Space root" + ) + diagnostics = self.git.diagnostics(root) + repository_id = ( + "repo_" + + canonical_digest({"space_id": space_id, "role": "content"})[ + :32 + ] + ) + ownership = ( + "eigent_owned" + if initialized and eigent_owned_space + else "adopted" + ) + repository = self.journal.put_git_repository( + repository_id=repository_id, + space_id=space_id, + repository_role="content", + root_path=str(root), + root_path_digest=canonical_digest(str(root)), + ownership=ownership, + state=("ready" if diagnostics.healthy else "needs_attention"), + version_coverage="managed_files_only", + hooks_mode="disabled", + repo_subdir=repo_subdir, + ) + return ContentRepositoryResult( + repository=repository, + probe=probe, + diagnostics=diagnostics, + initialized=initialized, + ) + + def status(self, repository_id: str) -> ContentRepositoryStatus: + repository = self._repository(repository_id) + diagnostics = self.git.diagnostics(Path(repository.root_path)) + repository = self._converge_repository_state( + repository, + diagnostics, + ) + return ContentRepositoryStatus( + repository=repository, + diagnostics=diagnostics, + managed_paths=self.journal.list_git_managed_paths(repository_id), + ) + + def diff( + self, + repository_id: str, + *, + paths: tuple[Path, ...], + source_commit: str | None = None, + ) -> str: + repository = self._repository(repository_id) + return self.git.diff_paths( + Path(repository.root_path), + paths, + source=source_commit, + ) + + def checkpoint( + self, + repository_id: str, + *, + operation_request_id: str, + expected_repo_state_digest: str, + paths: tuple[Path, ...], + path_sources: dict[str, str], + target_role: str, + target_id: str, + actor_id: str, + trigger: str, + message: str, + ) -> GitCheckpointRecord: + self._validate_identifier("operation_request_id", operation_request_id) + self._validate_text("actor_id", actor_id) + self._validate_text("trigger", trigger) + self._validate_text("message", message, max_length=500) + self._validate_text("target_id", target_id, max_length=256) + if target_role not in _TARGET_ROLES: + raise ValueError(f"unsupported checkpoint target {target_role!r}") + repository = self._repository(repository_id) + root = Path(repository.root_path) + self._assert_no_symlink_components(root, paths) + relative_paths = tuple(sorted(self.git.relative_paths(root, paths))) + self._validate_checkpoint_paths(paths, relative_paths) + if set(path_sources) != set(relative_paths): + raise ValueError( + "path_sources must use the exact repository-relative paths" + ) + invalid_sources = set(path_sources.values()) - _MANAGED_SOURCES + if invalid_sources: + raise ValueError( + "unsupported managed path source: " + + ", ".join(sorted(invalid_sources)) + ) + payload = { + "repository_id": repository_id, + "paths": list(relative_paths), + "path_sources": path_sources, + "target_role": target_role, + "target_id": target_id, + "actor_id": actor_id, + "trigger": trigger, + "message": message, + } + operation_id = ( + "gitop_" + + canonical_digest( + { + "repository_id": repository_id, + "request_id": operation_request_id, + } + )[:32] + ) + checkpoint_id = "checkpoint_" + operation_id.removeprefix("gitop_") + + with self._repository_lock(self._lock_path(repository.space_id)): + diagnostics = self.git.diagnostics(root) + repository = self._converge_repository_state( + repository, + diagnostics, + ) + if repository.state != "ready": + raise ContentRepositoryError( + f"repository {repository_id!r} is {repository.state!r}" + ) + operation = self.journal.begin_git_operation( + operation_id=operation_id, + repository_id=repository_id, + request_id=operation_request_id, + operation_type="checkpoint.create", + payload_digest=canonical_digest(payload), + expected_repo_state_digest=expected_repo_state_digest, + ) + if operation.status == "completed": + checkpoint = self.journal.get_git_checkpoint(checkpoint_id) + if checkpoint is None: + raise ContentRepositoryError( + "completed checkpoint operation has no checkpoint" + ) + return checkpoint + if operation.status == "dispatched": + recovered = self.git.find_commit_by_operation( + root, operation_id + ) + if recovered is not None: + return self._persist_checkpoint( + repository=repository, + operation_id=operation_id, + checkpoint_id=checkpoint_id, + commit_oid=recovered, + relative_paths=relative_paths, + path_sources=path_sources, + target_role=target_role, + target_id=target_id, + actor_id=actor_id, + trigger=trigger, + message=message, + ) + raise ContentRepositoryError( + "checkpoint outcome is unresolved and requires " + "reconciliation" + ) + current = self.git.repo_state_token(root) + if current.digest != expected_repo_state_digest: + self.journal.fail_git_operation( + operation_id, + error_code="repo_state_changed", + error_message="Repository changed before checkpoint", + ) + raise RepositoryStateChangedError( + "Repository changed before checkpoint; refresh status" + ) + changed = self.git.path_status(root, paths) + if not changed: + self.journal.fail_git_operation( + operation_id, + error_code="no_changes", + error_message="No selected path has a pending change", + ) + raise NoCheckpointChangesError( + "No selected path has a pending change" + ) + staged = sorted( + path + for path, state in changed.items() + if state != "??" and state[0] != " " + ) + if staged: + self.journal.fail_git_operation( + operation_id, + error_code="selected_paths_already_staged", + error_message=( + "Selected checkpoint paths already contain user-staged " + "changes" + ), + ) + raise ContentRepositoryError( + "Selected checkpoint paths already contain staged changes; " + "commit or unstage them before saving an Eigent checkpoint" + ) + parent_oid = current.head_oid + self.journal.mark_git_operation_dispatched( + operation_id, + observed_repo_state_digest=current.digest, + ) + commit_message = ( + f"{message}\n\n" + f"Eigent-Operation: {operation_id}\n" + f"Eigent-Actor: {actor_id}\n" + f"Eigent-Trigger: {trigger}" + ) + try: + commit_oid = self.git.commit_paths( + root, + paths, + message=commit_message, + author_name="Eigent User", + author_email="noreply@eigent.ai", + ) + except Exception as exc: + recovered = self.git.find_commit_by_operation( + root, operation_id + ) + if recovered is not None: + commit_oid = recovered + else: + after = self.git.repo_state_token(root) + self.journal.fail_git_operation( + operation_id, + error_code="git_checkpoint_failed", + error_message=str(exc)[:1000], + outcome_unknown=after.head_oid != parent_oid, + ) + raise + return self._persist_checkpoint( + repository=repository, + operation_id=operation_id, + checkpoint_id=checkpoint_id, + commit_oid=commit_oid, + relative_paths=relative_paths, + path_sources=path_sources, + target_role=target_role, + target_id=target_id, + actor_id=actor_id, + trigger=trigger, + message=message, + ) + + def prepare_restore_candidate( + self, + checkpoint_id: str, + *, + operation_request_id: str, + expected_repo_state_digest: str, + ) -> RestoreCandidate: + self._validate_identifier("operation_request_id", operation_request_id) + checkpoint = self.journal.get_git_checkpoint(checkpoint_id) + if checkpoint is None: + raise ContentRepositoryError( + f"unknown checkpoint {checkpoint_id!r}" + ) + repository = self._repository(checkpoint.repository_id) + root = Path(repository.root_path) + operation_id = ( + "gitop_" + + canonical_digest( + { + "repository_id": repository.repository_id, + "request_id": operation_request_id, + } + )[:32] + ) + ref_name = ( + f"refs/eigent/recovery/{repository.repository_id}/{operation_id}" + ) + payload = { + "checkpoint_id": checkpoint_id, + "ref_name": ref_name, + "commit_oid": checkpoint.commit_oid, + } + with self._repository_lock(self._lock_path(repository.space_id)): + operation = self.journal.begin_git_operation( + operation_id=operation_id, + repository_id=repository.repository_id, + request_id=operation_request_id, + operation_type="checkpoint.restore_candidate", + payload_digest=canonical_digest(payload), + expected_repo_state_digest=expected_repo_state_digest, + ) + if operation.status == "completed": + return RestoreCandidate( + operation=operation, + checkpoint=checkpoint, + ref_name=ref_name, + commit_oid=checkpoint.commit_oid, + ) + current = self.git.repo_state_token(root) + if operation.status == "dispatched": + existing_oid = self.git.ref_oid(root, ref_name) + if existing_oid == checkpoint.commit_oid: + operation = self.journal.complete_git_operation( + operation_id, + result=payload, + observed_repo_state_digest=current.digest, + ) + return RestoreCandidate( + operation=operation, + checkpoint=checkpoint, + ref_name=ref_name, + commit_oid=checkpoint.commit_oid, + ) + if current.digest != expected_repo_state_digest: + self.journal.fail_git_operation( + operation_id, + error_code="repo_state_changed", + error_message="Repository changed before restore preview", + ) + raise RepositoryStateChangedError( + "Repository changed before restore preview" + ) + existing_oid = self.git.ref_oid(root, ref_name) + if operation.status == "prepared": + self.journal.mark_git_operation_dispatched( + operation_id, + observed_repo_state_digest=current.digest, + ) + if existing_oid is None: + self.git.update_eigent_ref( + root, + ref_name, + checkpoint.commit_oid, + ) + elif existing_oid != checkpoint.commit_oid: + self.journal.fail_git_operation( + operation_id, + error_code="restore_ref_conflict", + error_message="Recovery ref points to a different commit", + outcome_unknown=True, + ) + raise ContentRepositoryError( + "Recovery ref conflicts with the requested checkpoint" + ) + observed = self.git.repo_state_token(root) + operation = self.journal.complete_git_operation( + operation_id, + result=payload, + observed_repo_state_digest=observed.digest, + ) + return RestoreCandidate( + operation=operation, + checkpoint=checkpoint, + ref_name=ref_name, + commit_oid=checkpoint.commit_oid, + ) + + def _persist_checkpoint( + self, + *, + repository: GitRepositoryRecord, + operation_id: str, + checkpoint_id: str, + commit_oid: str, + relative_paths: tuple[str, ...], + path_sources: dict[str, str], + target_role: str, + target_id: str, + actor_id: str, + trigger: str, + message: str, + ) -> GitCheckpointRecord: + root = Path(repository.root_path) + observed = self.git.repo_state_token(root) + return self.journal.complete_git_checkpoint( + checkpoint_id=checkpoint_id, + operation_id=operation_id, + repository_id=repository.repository_id, + target_role=target_role, + target_id=target_id, + commit_oid=commit_oid, + parent_oid=self.git.commit_parent(root, commit_oid), + paths=relative_paths, + managed_path_sources=path_sources, + actor_id=actor_id, + trigger=trigger, + message=message, + observed_repo_state_digest=observed.digest, + ) + + def _repository(self, repository_id: str) -> GitRepositoryRecord: + repository = self.journal.get_git_repository(repository_id) + if repository is None: + raise ContentRepositoryError( + f"unknown Content Repository {repository_id!r}" + ) + if repository.repository_role != "content": + raise ContentRepositoryError( + f"repository {repository_id!r} is not a Content Repository" + ) + return repository + + def _converge_repository_state( + self, + repository: GitRepositoryRecord, + diagnostics: RepositoryDiagnostics, + ) -> GitRepositoryRecord: + state = "ready" if diagnostics.healthy else "needs_attention" + if repository.state == state: + return repository + return self.journal.update_git_repository_state( + repository.repository_id, + state=state, + expected_version=repository.version, + ) + + @staticmethod + def _validate_identifier(name: str, value: str) -> None: + if not _SAFE_IDENTIFIER.fullmatch(value): + raise ValueError(f"invalid {name}") + + @staticmethod + def _validate_text( + name: str, + value: str, + *, + max_length: int = 200, + ) -> None: + if not value.strip() or len(value) > max_length or "\x00" in value: + raise ValueError(f"invalid {name}") + if "\r" in value or "\n" in value: + raise ValueError(f"{name} must be a single line") + + @staticmethod + def _validate_checkpoint_paths( + original_paths: tuple[Path, ...], + relative_paths: tuple[str, ...], + ) -> None: + if len(original_paths) != len(relative_paths) or len( + set(relative_paths) + ) != len(relative_paths): + raise ValueError("checkpoint paths must be unique") + for relative in relative_paths: + if relative == ".git" or relative.startswith(".git/"): + raise GitBackendError(".git cannot be checkpointed") + + @staticmethod + def _assert_no_symlink_components( + root: Path, + paths: tuple[Path, ...], + ) -> None: + for original in paths: + candidate = original.expanduser() + if not candidate.is_absolute(): + candidate = root / candidate + try: + lexical = candidate.absolute().relative_to(root) + except ValueError as exc: + raise GitBackendError( + f"checkpoint path escapes repository: {original}" + ) from exc + cursor = root + for part in lexical.parts: + cursor = cursor / part + if cursor.is_symlink(): + raise GitBackendError( + "symlink checkpoint path is not allowed: " + f"{lexical.as_posix()}" + ) + + def _lock_path(self, space_id: str) -> Path: + return ( + self.state_root + / "git-operation-locks" + / f"content-{space_id}.lock" + ) + + @contextmanager + def _repository_lock(self, lock_path: Path) -> Iterator[None]: + lock_path.parent.mkdir(parents=True, exist_ok=True) + with self._lock: + descriptor = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + try: + if fcntl is not None: + fcntl.flock(descriptor, fcntl.LOCK_EX) + yield + finally: + if fcntl is not None: + fcntl.flock(descriptor, fcntl.LOCK_UN) + os.close(descriptor) diff --git a/backend/tests/app/controller/test_workspace_git_controller.py b/backend/tests/app/controller/test_workspace_git_controller.py new file mode 100644 index 000000000..f3d33ca1b --- /dev/null +++ b/backend/tests/app/controller/test_workspace_git_controller.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.auth.local_control import LOCAL_CONTROL_CAPABILITY_HEADER +from app.controller import workspace_git_controller +from app.router import register_routers +from app.run_journal import SQLiteRunJournal +from app.workspace_git import ContentRepositoryService, GitBackend + + +@dataclass +class _Binding: + workspace_root: str + + +class _BindingStore: + def __init__(self, root: Path) -> None: + self.root = root + + def get_binding(self, _email, _space_id, _user_id): + return _Binding(workspace_root=str(self.root)) + + +class _Resolver: + def __init__(self, root: Path) -> None: + self.store = _BindingStore(root) + + +@pytest.fixture +def git_api(tmp_path, monkeypatch): + space = tmp_path / "space" + space.mkdir() + hooks = tmp_path / "empty-hooks" + hooks.mkdir() + journal = SQLiteRunJournal(tmp_path / "run-journal.sqlite3") + service = ContentRepositoryService( + journal, + state_root=tmp_path / "state", + git_backend=GitBackend(hooks_path=hooks), + ) + resolver = _Resolver(space) + monkeypatch.setattr(workspace_git_controller, "_service", lambda: service) + monkeypatch.setattr( + workspace_git_controller, + "get_workspace_resolver", + lambda: resolver, + ) + monkeypatch.setenv("EIGENT_RUNTIME", "electron") + monkeypatch.setenv("EIGENT_LOCAL_CONTROL_CAPABILITY", "test-secret") + app = FastAPI() + register_routers(app, prefix="/api/v1") + client = TestClient(app, client=("127.0.0.1", 50000)) + try: + yield client, service, resolver, space + finally: + client.close() + journal.close() + + +def _headers() -> dict[str, str]: + return {LOCAL_CONTROL_CAPABILITY_HEADER: "test-secret"} + + +def _status(client: TestClient) -> dict: + response = client.get( + "/api/v1/spaces/space-1/git/status", + params={"email": "user@example.com"}, + headers=_headers(), + ) + assert response.status_code == 200 + return response.json() + + +def test_workspace_git_api_requires_local_renderer_capability(git_api): + client, _, _, _ = git_api + + response = client.get( + "/api/v1/spaces/space-1/git/status", + params={"email": "user@example.com"}, + ) + + assert response.status_code == 401 + + +def test_workspace_git_checkpoint_and_restore_candidate_flow(git_api): + client, _, _, space = git_api + report = space / "report.md" + report.write_text("first version\n", encoding="utf-8") + + inspection = _status(client) + assert inspection["enabled"] is False + assert inspection["consent_required"] is True + assert str(space) not in str(inspection) + + response = client.post( + "/api/v1/spaces/space-1/git/bootstrap", + headers=_headers(), + json={ + "email": "user@example.com", + "allow_init": False, + }, + ) + assert response.status_code == 409 + assert response.json()["detail"]["code"] == "git_init_consent_required" + + response = client.post( + "/api/v1/spaces/space-1/git/bootstrap", + headers=_headers(), + json={ + "email": "user@example.com", + "allow_init": True, + }, + ) + assert response.status_code == 200 + assert response.json()["initialized"] is True + assert str(space) not in response.text + + status = _status(client) + state_digest = status["diagnostics"]["repo_state"]["digest"] + response = client.post( + "/api/v1/spaces/space-1/git/checkpoints", + headers=_headers(), + json={ + "email": "user@example.com", + "operation_request_id": "save-1", + "expected_repo_state_digest": state_digest, + "paths": ["report.md"], + "path_sources": {"report.md": "agent_created"}, + "target_role": "user", + "target_id": "space-1", + "actor_id": "user-1", + "trigger": "user_save", + "message": "Save progress", + }, + ) + assert response.status_code == 201 + checkpoint_id = response.json()["checkpoint_id"] + assert response.json()["paths"] == ["report.md"] + + response = client.get( + "/api/v1/spaces/space-1/git/checkpoints", + params={"email": "user@example.com"}, + headers=_headers(), + ) + assert response.status_code == 200 + assert response.json()["checkpoints"][0]["checkpoint_id"] == checkpoint_id + assert str(space) not in response.text + + report.write_text("second version\n", encoding="utf-8") + response = client.get( + "/api/v1/spaces/space-1/git/diff", + params={ + "email": "user@example.com", + "paths": "report.md", + }, + headers=_headers(), + ) + assert response.status_code == 200 + assert "+second version" in response.json()["diff"] + + status = _status(client) + response = client.post( + "/api/v1/spaces/space-1/git/restore", + headers=_headers(), + json={ + "email": "user@example.com", + "checkpoint_id": checkpoint_id, + "operation_request_id": "restore-1", + "expected_repo_state_digest": status["diagnostics"]["repo_state"][ + "digest" + ], + }, + ) + assert response.status_code == 201 + assert response.json()["checkpoint_id"] == checkpoint_id + assert response.json()["candidate_ref"].startswith("refs/eigent/recovery/") + assert response.json()["applied_to_user_worktree"] is False + assert report.read_text(encoding="utf-8") == "second version\n" + + +def test_workspace_git_status_fails_closed_after_space_rebind(git_api): + client, _, resolver, _ = git_api + response = client.post( + "/api/v1/spaces/space-1/git/bootstrap", + headers=_headers(), + json={ + "email": "user@example.com", + "allow_init": True, + }, + ) + assert response.status_code == 200 + rebound = resolver.store.root.parent / "rebound" + rebound.mkdir() + resolver.store.root = rebound + + response = client.get( + "/api/v1/spaces/space-1/git/status", + params={"email": "user@example.com"}, + headers=_headers(), + ) + + assert response.status_code == 409 + assert response.json()["detail"]["code"] == ( + "git_repository_binding_mismatch" + ) + + +def test_workspace_git_rejects_escaping_checkpoint_path(git_api): + client, _, _, _ = git_api + + response = client.post( + "/api/v1/spaces/space-1/git/checkpoints", + headers=_headers(), + json={ + "email": "user@example.com", + "operation_request_id": "save-escape", + "expected_repo_state_digest": "0" * 64, + "paths": ["../secret.txt"], + "path_sources": {"../secret.txt": "user_selected"}, + "target_role": "user", + "target_id": "space-1", + "actor_id": "user-1", + "trigger": "user_save", + "message": "Invalid save", + }, + ) + + assert response.status_code == 422 diff --git a/backend/tests/app/workspace_git/test_content_repository.py b/backend/tests/app/workspace_git/test_content_repository.py new file mode 100644 index 000000000..f5ac7a2f3 --- /dev/null +++ b/backend/tests/app/workspace_git/test_content_repository.py @@ -0,0 +1,488 @@ +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest + +from app.run_journal import IdempotencyConflictError, SQLiteRunJournal +from app.workspace_config import canonical_digest +from app.workspace_git import ( + ContentRepositoryConsentRequired, + ContentRepositoryError, + ContentRepositoryService, + GitBackend, + GitBackendError, + NestedRepositoryError, + RepositoryStateChangedError, +) + + +@pytest.fixture +def journal(tmp_path): + with SQLiteRunJournal(tmp_path / "run-journal.sqlite3") as value: + yield value + + +def _service(tmp_path: Path, journal: SQLiteRunJournal): + hooks = tmp_path / "empty-hooks" + hooks.mkdir(exist_ok=True) + backend = GitBackend(hooks_path=hooks) + return ( + ContentRepositoryService( + journal, + state_root=tmp_path / "state", + git_backend=backend, + ), + backend, + ) + + +def _git(repository: Path, *args: str, check: bool = True) -> str: + completed = subprocess.run( + ("git", "-C", str(repository), *args), + check=check, + capture_output=True, + text=True, + env={ + "PATH": "/usr/bin:/bin:/usr/local/bin:/opt/homebrew/bin", + "GIT_CONFIG_NOSYSTEM": "1", + "GIT_CONFIG_GLOBAL": "/dev/null", + "GIT_TERMINAL_PROMPT": "0", + }, + ) + return completed.stdout.strip() + + +def test_plain_folder_inspection_is_read_only_and_requires_consent( + tmp_path, + journal, +): + space = tmp_path / "user-folder" + space.mkdir() + private = space / "private.txt" + private.write_text("do not stage", encoding="utf-8") + service, _ = _service(tmp_path, journal) + + inspection = service.inspect(space) + + assert inspection.enablement == "not_enabled" + assert inspection.consent_required is True + assert not (space / ".git").exists() + assert private.read_text(encoding="utf-8") == "do not stage" + assert journal.get_space_git_repository(space_id="space-1") is None + + +def test_user_folder_init_requires_consent_and_never_auto_stages_files( + tmp_path, + journal, +): + space = tmp_path / "user-folder" + space.mkdir() + private = space / "private.txt" + private.write_text("do not stage", encoding="utf-8") + service, _ = _service(tmp_path, journal) + + with pytest.raises(ContentRepositoryConsentRequired): + service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=False, + ) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=True, + ) + + assert result.initialized is True + assert result.repository.ownership == "adopted" + assert result.repository.version_coverage == "managed_files_only" + assert _git(space, "status", "--porcelain") == "?? private.txt" + assert _git(space, "rev-parse", "--verify", "HEAD", check=False) == "" + + +def test_adopt_preserves_branch_remote_and_repository_config( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + backend.init_repository(space, initial_branch="trunk") + seed = space / "seed.txt" + seed.write_text("seed", encoding="utf-8") + backend.commit_paths(space, (seed,), message="seed") + _git(space, "remote", "add", "origin", "https://example.invalid/repo") + before_config = (space / ".git" / "config").read_bytes() + + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=False, + ) + + assert result.initialized is False + assert result.repository.ownership == "adopted" + assert result.probe.branch == "trunk" + assert result.diagnostics.has_remotes is True + assert (space / ".git" / "config").read_bytes() == before_config + + +def test_nested_repository_requires_explicit_parent_binding( + tmp_path, + journal, +): + parent = tmp_path / "parent" + parent.mkdir() + child = parent / "child" + child.mkdir() + service, backend = _service(tmp_path, journal) + backend.init_repository(parent) + + inspection = service.inspect(child) + assert inspection.enablement == "nested_repository_requires_binding" + with pytest.raises(NestedRepositoryError): + service.bootstrap( + space_id="space-1", + space_root=child, + allow_init=True, + repo_subdir="child", + ) + assert not (child / ".git").exists() + + +def test_checkpoint_commits_only_explicit_paths_and_preserves_user_index( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + backend.init_repository(space) + first = space / "first.txt" + unrelated = space / "unrelated.txt" + first.write_text("baseline", encoding="utf-8") + unrelated.write_text("baseline", encoding="utf-8") + backend.commit_paths(space, (first, unrelated), message="baseline") + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=False, + ) + + first.write_text("agent edit", encoding="utf-8") + unrelated.write_text("user staged edit", encoding="utf-8") + _git(space, "add", "--", "unrelated.txt") + expected = backend.repo_state_token(space) + checkpoint = service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-1", + expected_repo_state_digest=expected.digest, + paths=(first,), + path_sources={"first.txt": "agent_modified"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Save progress", + ) + replay = service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-1", + expected_repo_state_digest=expected.digest, + paths=(first,), + path_sources={"first.txt": "agent_modified"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Save progress", + ) + + assert checkpoint == replay + assert checkpoint.paths == ("first.txt",) + assert backend.show_commit_paths(space, checkpoint.commit_oid) == ( + "first.txt", + ) + assert _git(space, "diff", "--cached", "--name-only") == ("unrelated.txt") + assert journal.list_git_managed_paths(result.repository.repository_id) == ( + "first.txt", + ) + + with pytest.raises(IdempotencyConflictError): + service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-1", + expected_repo_state_digest=expected.digest, + paths=(unrelated,), + path_sources={"unrelated.txt": "user_selected"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Different payload", + ) + + +def test_checkpoint_refuses_to_overwrite_selected_path_staging( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + backend.init_repository(space) + target = space / "report.md" + target.write_text("baseline\n", encoding="utf-8") + backend.commit_paths(space, (target,), message="baseline") + head_before = backend.current_head(space) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=False, + ) + target.write_text("user staged version\n", encoding="utf-8") + _git(space, "add", "--", "report.md") + staged_blob_before = _git(space, "rev-parse", ":report.md") + target.write_text("working version\n", encoding="utf-8") + expected = backend.repo_state_token(space) + + with pytest.raises(ContentRepositoryError, match="staged changes"): + service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-staged-target", + expected_repo_state_digest=expected.digest, + paths=(target,), + path_sources={"report.md": "user_selected"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Save progress", + ) + + assert _git(space, "rev-parse", ":report.md") == staged_blob_before + assert target.read_text(encoding="utf-8") == "working version\n" + assert backend.current_head(space) == head_before + + +def test_checkpoint_repo_state_cas_rejects_external_change( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=True, + ) + target = space / "report.md" + target.write_text("v1", encoding="utf-8") + expected = backend.repo_state_token(space) + (space / "external.txt").write_text("changed", encoding="utf-8") + + with pytest.raises(RepositoryStateChangedError): + service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-cas", + expected_repo_state_digest=expected.digest, + paths=(target,), + path_sources={"report.md": "agent_created"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Save progress", + ) + + operation_id = ( + "gitop_" + + canonical_digest( + { + "repository_id": result.repository.repository_id, + "request_id": "checkpoint-cas", + } + )[:32] + ) + assert journal.get_git_operation(operation_id).status == "failed" + assert journal.get_git_operation(operation_id).error_code == ( + "repo_state_changed" + ) + + +def test_repository_state_recovers_after_transient_attention_state( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, _ = _service(tmp_path, journal) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=True, + ) + attention = journal.update_git_repository_state( + result.repository.repository_id, + state="needs_attention", + expected_version=result.repository.version, + ) + + status = service.status(result.repository.repository_id) + + assert attention.state == "needs_attention" + assert status.repository.state == "ready" + assert status.repository.version == attention.version + 1 + + +def test_checkpoint_recovery_closes_commit_before_journal_window( + tmp_path, + journal, + monkeypatch, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=True, + ) + target = space / "result.md" + target.write_text("done", encoding="utf-8") + expected = backend.repo_state_token(space) + original = journal.complete_git_checkpoint + + def crash_before_journal(*_args, **_kwargs): + raise RuntimeError("simulated crash after git commit") + + monkeypatch.setattr( + journal, "complete_git_checkpoint", crash_before_journal + ) + with pytest.raises(RuntimeError, match="simulated crash"): + service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-crash", + expected_repo_state_digest=expected.digest, + paths=(target,), + path_sources={"result.md": "agent_created"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="run_terminal", + message="Run checkpoint", + ) + monkeypatch.setattr(journal, "complete_git_checkpoint", original) + + recovered = service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-crash", + expected_repo_state_digest=expected.digest, + paths=(target,), + path_sources={"result.md": "agent_created"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="run_terminal", + message="Run checkpoint", + ) + + assert journal.get_git_operation(recovered.operation_id).status == ( + "completed" + ) + assert ( + backend.find_commit_by_operation(space, recovered.operation_id) + == recovered.commit_oid + ) + + +def test_checkpoint_rejects_repository_clean_filter_without_execution( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=True, + ) + marker = tmp_path / "filter-executed" + attributes = space / ".gitattributes" + attributes.write_text("danger.txt filter=pwn\n", encoding="utf-8") + danger = space / "danger.txt" + danger.write_text("secret", encoding="utf-8") + _git( + space, + "config", + "filter.pwn.clean", + f"touch {marker}", + ) + expected = backend.repo_state_token(space) + + with pytest.raises(GitBackendError, match="clean filter"): + service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-filter", + expected_repo_state_digest=expected.digest, + paths=(danger,), + path_sources={"danger.txt": "agent_created"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Unsafe filter test", + ) + + assert not marker.exists() + + +def test_restore_candidate_only_creates_private_ref( + tmp_path, + journal, +): + space = tmp_path / "repo" + space.mkdir() + service, backend = _service(tmp_path, journal) + result = service.bootstrap( + space_id="space-1", + space_root=space, + allow_init=True, + ) + target = space / "report.md" + target.write_text("checkpoint", encoding="utf-8") + checkpoint = service.checkpoint( + result.repository.repository_id, + operation_request_id="checkpoint-restore-source", + expected_repo_state_digest=backend.repo_state_token(space).digest, + paths=(target,), + path_sources={"report.md": "agent_created"}, + target_role="user", + target_id="space-1", + actor_id="user-1", + trigger="user_save", + message="Restore source", + ) + target.write_text("new working edit", encoding="utf-8") + expected = backend.repo_state_token(space) + head_before = backend.current_head(space) + + candidate = service.prepare_restore_candidate( + checkpoint.checkpoint_id, + operation_request_id="restore-1", + expected_repo_state_digest=expected.digest, + ) + replay = service.prepare_restore_candidate( + checkpoint.checkpoint_id, + operation_request_id="restore-1", + expected_repo_state_digest=expected.digest, + ) + + assert candidate == replay + assert backend.current_head(space) == head_before + assert target.read_text(encoding="utf-8") == "new working edit" + assert backend.ref_oid(space, candidate.ref_name) == checkpoint.commit_oid