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

feat: add durable content repository checkpoints

Browse source
This commit is contained in:
4pmtong 2026-08-07 17:18:14 +08:00
parent 0fd7b99641
commit 5cb411d9ba
10 changed files with 2952 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

433
backend/app/controller/workspace_git_controller.py Normal file
View file

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

7
backend/app/router.py
View file

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

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

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

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

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

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

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

24
backend/app/workspace_git/__init__.py
View file

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

259
backend/app/workspace_git/backend.py
View file

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

679
backend/app/workspace_git/content.py Normal file
View file

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

234
backend/tests/app/controller/test_workspace_git_controller.py Normal file
View file

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

488
backend/tests/app/workspace_git/test_content_repository.py Normal file
View file

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