# ========= 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, nullcontext 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, ...] pending_managed_paths: tuple[str, ...] pending_managed_paths_truncated: bool @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.repository_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 describes the Space, not whether this particular # request happened to run `git init`. A retry after init but before # the SQLite binding must preserve an Eigent-created blank Space. ownership = "eigent_owned" if 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) root = Path(repository.root_path) diagnostics = self.git.diagnostics(root) repository = self._converge_repository_state( repository, diagnostics, ) managed_paths = self.journal.list_git_managed_paths(repository_id) pending_managed_paths, pending_truncated = self._pending_managed_paths( root, managed_paths, limit=500, ) return ContentRepositoryStatus( repository=repository, diagnostics=diagnostics, managed_paths=managed_paths, pending_managed_paths=pending_managed_paths, pending_managed_paths_truncated=pending_truncated, ) def _pending_managed_paths( self, root: Path, managed_paths: tuple[str, ...], *, limit: int, ) -> tuple[tuple[str, ...], bool]: """Return only managed deltas without scanning unrelated files.""" if limit < 1: raise ValueError("pending managed path limit must be positive") pending: list[str] = [] chunk_size = 500 for offset in range(0, len(managed_paths), chunk_size): chunk = managed_paths[offset : offset + chunk_size] changed = self.git.path_status( root, tuple(root / value for value in chunk), ) for relative_path in chunk: if relative_path not in changed: continue pending.append(relative_path) if len(pending) > limit: return tuple(pending[:limit]), True return tuple(pending), False 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, worktree_root: Path | None = None, repository_lock_held: bool = False, commit_trailers: dict[str, str] | None = None, ) -> 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) trailers = dict(commit_trailers or {}) for key, value in trailers.items(): if not re.fullmatch(r"[A-Za-z][A-Za-z0-9-]{0,63}", key): raise ValueError("invalid commit trailer key") self._validate_text("commit trailer value", value, 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 = self._checkpoint_root(repository, worktree_root) 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, "commit_trailers": trailers, "worktree_ref": self._worktree_ref(repository, root), } operation_id = ( "gitop_" + canonical_digest( { "repository_id": repository_id, "request_id": operation_request_id, } )[:32] ) checkpoint_id = "checkpoint_" + operation_id.removeprefix("gitop_") lock = ( nullcontext() if repository_lock_held else self.repository_lock( self.repository_lock_path(repository.space_id) ) ) with lock: 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, root=root, ) 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}" ) if trailers: commit_message += "\n" + "\n".join( f"{key}: {value}" for key, value in sorted(trailers.items()) ) 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, root=root, ) 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.repository_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, root: Path, ) -> GitCheckpointRecord: 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 _checkpoint_root( self, repository: GitRepositoryRecord, requested: Path | None, ) -> Path: user_root = Path(repository.root_path).expanduser().resolve() if requested is None: return user_root target = requested.expanduser().resolve() for worktree in self.git.list_worktrees(user_root): if worktree.path != target: continue if target == user_root: return target if worktree.ref_name and worktree.ref_name.startswith( "refs/heads/eigent/" ): return target raise ContentRepositoryError( "checkpoint target is not an Eigent-owned worktree" ) raise ContentRepositoryError( "checkpoint target is not registered with the Content Repository" ) def _worktree_ref( self, repository: GitRepositoryRecord, root: Path, ) -> str: user_root = Path(repository.root_path).expanduser().resolve() if root == user_root: return "USER_WORKTREE" for worktree in self.git.list_worktrees(user_root): if worktree.path == root and worktree.ref_name: return worktree.ref_name raise ContentRepositoryError("checkpoint worktree ref is unavailable") 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 repository_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)