# ========= 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. ========= """Pre-dispatch Run worktree materialization and exact-path ChangeSets.""" from __future__ import annotations import hashlib import logging import os import uuid from dataclasses import dataclass from pathlib import Path from app.run_context import RunContext from app.run_journal import ( GitChangeSetRecord, GitMutationIntentRecord, ProjectWorkspaceBindingRecord, SQLiteRunJournal, configured_run_journal_path, get_default_run_journal, ) from app.workspace_config import canonical_digest from app.workspace_git.content import ContentRepositoryError from app.workspace_git.coordinator import ( GitRunWorkspace, WorkspaceGitCoordinator, ) from app.workspace_git.snapshot import ( MaterializedOverlay, WorkspacePathNotFoundError, WorkspaceSnapshotService, ) from app.workspace_git.workforce import ( GitAgentWorkspace, WorkforceGitService, ) @dataclass(frozen=True) class PreparedWorkspaceWrite: context: RunContext workspace: GitRunWorkspace | None agent_workspace: GitAgentWorkspace | None change_set: GitChangeSetRecord intent: GitMutationIntentRecord relative_path: str target_path: Path preimage_digest: str | None overlay: MaterializedOverlay | None direct_binding: ProjectWorkspaceBindingRecord | None = None @property def mutation_root(self) -> Path: if self.direct_binding is not None: return Path(self.direct_binding.worktree_path) if self.agent_workspace is None: raise ContentRepositoryError("prepared write has no mutation root") return self.agent_workspace.agent_worktree @dataclass(frozen=True) class PreparedWorkspaceExecution: context: RunContext workspace: GitRunWorkspace | None agent_workspace: GitAgentWorkspace | None change_set: GitChangeSetRecord intent: GitMutationIntentRecord imported_overlays: tuple[MaterializedOverlay, ...] direct_binding: ProjectWorkspaceBindingRecord | None = None @property def mutation_root(self) -> Path: if self.direct_binding is not None: return Path(self.direct_binding.worktree_path) if self.agent_workspace is None: raise ContentRepositoryError( "prepared execution has no mutation root" ) return self.agent_workspace.agent_worktree @dataclass(frozen=True) class WorkspaceMutationReconciliation: recovered_change_set_ids: tuple[str, ...] needs_attention_change_set_ids: tuple[str, ...] logger = logging.getLogger(__name__) class WorkspaceMutationService: """Enforce materialize-before-dispatch for workspace mutations.""" def __init__( self, journal: SQLiteRunJournal, *, state_root: Path, coordinator: WorkspaceGitCoordinator | None = None, snapshots: WorkspaceSnapshotService | None = None, workforce: WorkforceGitService | None = None, primary_checkout_enabled: bool = True, ) -> None: self.journal = journal self.state_root = state_root.expanduser().resolve() self.coordinator = coordinator or WorkspaceGitCoordinator( journal, state_root=self.state_root, ) self.snapshots = snapshots or WorkspaceSnapshotService( journal, state_root=self.state_root, git_backend=self.coordinator.git, ) self.content = self.coordinator.content self.git = self.coordinator.git self.workforce = workforce or WorkforceGitService( journal, state_root=self.state_root, coordinator=self.coordinator, ) self.primary_checkout_enabled = primary_checkout_enabled def prepare_file_write( self, *, context: RunContext, filename: str, operation_request_id: str, actor_id: str, trigger: str, ) -> PreparedWorkspaceWrite | None: """Return a private Run target, or None when Git is not enabled.""" run = self.journal.get_run_git_materialization(context.run_id) if run is None: return None project = self.journal.get_project_git_state(run.project_id) repository = self.journal.get_git_repository(run.repository_id) if project is None or repository is None: raise ContentRepositoryError("Run Git owner is unavailable") relative_path = self._relative_workspace_path( context=context, repository_root=Path(repository.root_path), filename=filename, ) binding = self.journal.get_project_workspace_binding( context.project_id ) if ( self.primary_checkout_enabled and context.session_mode == "single-agent" and binding is not None and binding.checkout_mode in {"primary_checkout", "explicit_worktree"} ): return self._prepare_direct_file_write( context=context, run=run, repository_root=Path(repository.root_path), binding=binding, relative_path=relative_path, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, ) pinned_entry = None try: pinned_entry = self.snapshots.pin_path( run_id=context.run_id, relative_path=relative_path, ) except WorkspacePathNotFoundError: pass user_token = self.git.repo_state_token(Path(repository.root_path)) workspace = self.coordinator.ensure_run_materialized( run_id=context.run_id, operation_request_id=self._request_id( operation_request_id, "materialize", ), expected_repo_state_digest=user_token.digest, expected_project_version=project.version, expected_project_head=project.integration_head, ) agent_workspace = self.workforce.ensure_agent_workspace( run_workspace=workspace, agent_id=actor_id, operation_request_id=operation_request_id, ) target = agent_workspace.agent_worktree / relative_path overlay = None if ( pinned_entry is not None and pinned_entry.source_kind == "user_overlay" and pinned_entry.entry_state in {"read_only", "imported_preimage"} ): overlay = self.snapshots.materialize_user_overlay( run_id=context.run_id, relative_path=relative_path, destination_root=agent_workspace.agent_worktree, ) preimage_digest = ( overlay.content_digest if overlay is not None else self._digest_file(target) ) change_set = self.journal.ensure_git_change_set( change_set_id=( "changeset_" + canonical_digest( { "run_id": context.run_id, "worktree_ref": agent_workspace.record.agent_ref, } )[:32] ), run_id=context.run_id, repository_id=run.repository_id, worktree_ref=agent_workspace.record.agent_ref, base_commit=agent_workspace.record.base_commit, ) existing_intent = next( ( item for item in self.journal.list_git_mutation_intents() if item.change_set_id == change_set.change_set_id and item.operation_request_id == operation_request_id ), None, ) if existing_intent is not None: # A retry after checkpoint-but-before-merge observes the Agent # result as current content. Its durable intent must retain the # original preimage so idempotency and ChangeSet identity do not # drift with the worktree projection. preimage_digest = existing_intent.preimage_digest intent = self.journal.ensure_git_mutation_intent( intent_id=self._intent_id( change_set.change_set_id, operation_request_id, ), change_set_id=change_set.change_set_id, operation_request_id=operation_request_id, mutation_scope="exact_path", relative_path=relative_path, preimage_digest=preimage_digest, actor_id=actor_id, trigger=trigger, ) return PreparedWorkspaceWrite( context=context, workspace=workspace, agent_workspace=agent_workspace, change_set=change_set, intent=intent, relative_path=relative_path, target_path=target, preimage_digest=preimage_digest, overlay=overlay, ) def prepare_broad_write( self, *, context: RunContext, operation_request_id: str, actor_id: str, trigger: str, ) -> PreparedWorkspaceExecution | None: """Materialize before spawning terminal/script-like processes. Only overlay paths already admitted to the Run snapshot are imported. Expanding the import scope is a separate permission decision; this method never scans or copies the entire User Worktree. """ run = self.journal.get_run_git_materialization(context.run_id) if run is None: return None project = self.journal.get_project_git_state(run.project_id) repository = self.journal.get_git_repository(run.repository_id) if project is None or repository is None: raise ContentRepositoryError("Run Git owner is unavailable") binding = self.journal.get_project_workspace_binding( context.project_id ) if ( self.primary_checkout_enabled and context.session_mode == "single-agent" and binding is not None and binding.checkout_mode in {"primary_checkout", "explicit_worktree"} ): return self._prepare_direct_broad_write( context=context, run=run, repository_root=Path(repository.root_path), binding=binding, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, ) user_token = self.git.repo_state_token(Path(repository.root_path)) workspace = self.coordinator.ensure_run_materialized( run_id=context.run_id, operation_request_id=self._request_id( operation_request_id, "materialize", ), expected_repo_state_digest=user_token.digest, expected_project_version=project.version, expected_project_head=project.integration_head, ) agent_workspace = self.workforce.ensure_agent_workspace( run_workspace=workspace, agent_id=actor_id, operation_request_id=operation_request_id, ) change_set = self.journal.ensure_git_change_set( change_set_id=( "changeset_" + canonical_digest( { "run_id": context.run_id, "worktree_ref": agent_workspace.record.agent_ref, } )[:32] ), run_id=context.run_id, repository_id=run.repository_id, worktree_ref=agent_workspace.record.agent_ref, base_commit=agent_workspace.record.base_commit, ) intent = self.journal.ensure_git_mutation_intent( intent_id=self._intent_id( change_set.change_set_id, operation_request_id, ), change_set_id=change_set.change_set_id, operation_request_id=operation_request_id, mutation_scope="broad_process", relative_path=None, preimage_digest=None, actor_id=actor_id, trigger=trigger, ) imported: list[MaterializedOverlay] = [] snapshot = self.snapshots.get_snapshot(context.run_id) if snapshot is not None: for entry in self.journal.list_workspace_overlay_entries( snapshot.snapshot_id ): if entry.source_kind != "user_overlay": continue if entry.entry_state in {"read_only", "imported_preimage"}: imported.append( self.snapshots.materialize_user_overlay( run_id=context.run_id, relative_path=entry.relative_path, destination_root=agent_workspace.agent_worktree, ) ) return PreparedWorkspaceExecution( context=context, workspace=workspace, agent_workspace=agent_workspace, change_set=change_set, intent=intent, imported_overlays=tuple(imported), ) def complete_broad_write( self, prepared: PreparedWorkspaceExecution, *, operation_request_id: str, actor_id: str, trigger: str, ) -> tuple[str, ...]: """Capture a bounded terminal/script delta after the process exits.""" if prepared.direct_binding is not None: return self._complete_direct_broad_write( prepared, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, ) if prepared.agent_workspace is None: raise ContentRepositoryError( "legacy broad write has no Agent worktree" ) commits: list[str] = [] overlay_paths = { item.relative_path: item for item in prepared.imported_overlays } mutation_root = prepared.agent_workspace.agent_worktree status = self.git.worktree_status(mutation_root) for relative_path, overlay in overlay_paths.items(): if relative_path not in status: continue current_digest = self._digest_file(mutation_root / relative_path) if current_digest == overlay.content_digest: self.git.restore_owned_worktree_path( mutation_root, relative_path=relative_path, source_commit=prepared.agent_workspace.record.base_commit, ) continue commit = self.complete_file_write( PreparedWorkspaceWrite( context=prepared.context, workspace=prepared.workspace, agent_workspace=prepared.agent_workspace, change_set=prepared.change_set, intent=prepared.intent, relative_path=relative_path, target_path=mutation_root / relative_path, preimage_digest=overlay.content_digest, overlay=overlay, ), operation_request_id=self._request_id( operation_request_id, f"overlay:{relative_path}", ), actor_id=actor_id, trigger=trigger, merge_after_checkpoint=False, ) if commit is not None: commits.append(commit) status = self.git.worktree_status(mutation_root) remaining = { path: state for path, state in status.items() if path not in overlay_paths } if not remaining: self._complete_intent(prepared.intent) if commits: outcome = self.workforce.merge_agent_workspace( prepared.agent_workspace, operation_request_id=self._request_id( operation_request_id, "merge-agent", ), ) if outcome.merged_commit is not None: commits = [outcome.merged_commit] else: self.workforce.release_workspace(prepared.agent_workspace) return tuple(commits) paths: list[Path] = [] sources: dict[str, str] = {} pending_items: list[str] = [] for relative_path in sorted(remaining): target = mutation_root / relative_path tracked = self.git.is_tracked( mutation_root, target, ) result_digest = self._digest_file(target) if result_digest is None: change_kind = "deleted" size = None else: change_kind = "modified" if tracked else "added" size = target.stat().st_size source = "agent_modified" if tracked else "agent_created" self.journal.put_git_change_set_item( change_set_id=prepared.change_set.change_set_id, relative_path=relative_path, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, change_kind=change_kind, source="worktree_delta", preimage_digest=None, result_digest=result_digest, size_bytes=size, ) paths.append(target) sources[relative_path] = source pending_items.append(relative_path) checkpoint = self.content.checkpoint( prepared.change_set.repository_id, operation_request_id=self._request_id( operation_request_id, "terminal-delta", ), expected_repo_state_digest=self.git.repo_state_token( mutation_root ).digest, paths=tuple(paths), path_sources=sources, target_role="agent", target_id=prepared.agent_workspace.record.workspace_id, actor_id=actor_id, trigger=trigger, message="Checkpoint bounded workspace process delta", worktree_root=mutation_root, ) for relative_path in pending_items: self.journal.update_git_change_set_item_state( change_set_id=prepared.change_set.change_set_id, relative_path=relative_path, expected_state="pending", state="checkpointed", ) commits.append(checkpoint.commit_oid) self._complete_intent(prepared.intent) outcome = self.workforce.merge_agent_workspace( prepared.agent_workspace, operation_request_id=self._request_id( operation_request_id, "merge-agent", ), ) if outcome.merged_commit is not None: commits = [outcome.merged_commit] return tuple(commits) def renew_broad_write( self, prepared: PreparedWorkspaceExecution, *, now: float | None = None, ) -> None: """Renew the lease while an admitted broad process is still running.""" if prepared.direct_binding is not None: return if prepared.agent_workspace is None: raise ContentRepositoryError( "legacy broad write has no Agent worktree" ) self.workforce.renew_workspace( prepared.agent_workspace, now=now, ) def complete_file_write( self, prepared: PreparedWorkspaceWrite, *, operation_request_id: str, actor_id: str, trigger: str, merge_after_checkpoint: bool = True, ) -> str | None: """Checkpoint one successful exact-path write and return its commit.""" if prepared.direct_binding is not None: return self._complete_direct_file_write( prepared, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, ) if prepared.agent_workspace is None: raise ContentRepositoryError( "legacy file write has no Agent worktree" ) result_digest = self._digest_file(prepared.target_path) if result_digest is None and prepared.preimage_digest is None: self._complete_intent(prepared.intent) if merge_after_checkpoint: self.workforce.release_workspace(prepared.agent_workspace) return None if result_digest == prepared.preimage_digest: if prepared.overlay is not None: self.git.restore_owned_worktree_path( prepared.agent_workspace.agent_worktree, relative_path=prepared.relative_path, source_commit=prepared.agent_workspace.record.base_commit, ) self._complete_intent(prepared.intent) if merge_after_checkpoint: self.workforce.release_workspace(prepared.agent_workspace) return None size = ( prepared.target_path.stat().st_size if result_digest is not None else None ) change_kind = ( "deleted" if result_digest is None else ("added" if prepared.preimage_digest is None else "modified") ) source = ( "agent_created" if change_kind == "added" else "agent_modified" ) item = self.journal.put_git_change_set_item( change_set_id=prepared.change_set.change_set_id, relative_path=prepared.relative_path, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, change_kind=change_kind, source=source, preimage_digest=prepared.preimage_digest, result_digest=result_digest, size_bytes=size, ) if item.item_state == "checkpointed": self._complete_intent(prepared.intent) if merge_after_checkpoint: outcome = self.workforce.merge_agent_workspace( prepared.agent_workspace, operation_request_id=self._request_id( operation_request_id, "merge-agent", ), ) return outcome.merged_commit or self.git.current_head( prepared.agent_workspace.agent_worktree ) return self.git.current_head( prepared.agent_workspace.agent_worktree ) if prepared.overlay is not None and item.item_state == "pending": result_cache = ( self._cache_result( prepared.target_path, expected_digest=result_digest, ) if result_digest is not None else None ) self._replace_from_file( prepared.overlay.preimage_path, prepared.target_path, ) preimage_checkpoint = self.content.checkpoint( prepared.change_set.repository_id, operation_request_id=self._request_id( operation_request_id, "overlay-preimage", ), expected_repo_state_digest=self.git.repo_state_token( prepared.agent_workspace.agent_worktree ).digest, paths=(prepared.target_path,), path_sources={prepared.relative_path: "overlay_preimage"}, target_role="agent", target_id=prepared.agent_workspace.record.workspace_id, actor_id="user", trigger="overlay_preimage", message=( f"Preserve User preimage for {prepared.relative_path}" ), worktree_root=prepared.agent_workspace.agent_worktree, ) del preimage_checkpoint self.journal.update_git_change_set_item_state( change_set_id=prepared.change_set.change_set_id, relative_path=prepared.relative_path, expected_state="pending", state="preimage_checkpointed", ) if result_cache is not None: self._replace_from_file(result_cache, prepared.target_path) else: prepared.target_path.unlink(missing_ok=True) item = next( value for value in self.journal.list_git_change_set_items( prepared.change_set.change_set_id, ) if value.relative_path == prepared.relative_path ) checkpoint = self.content.checkpoint( prepared.change_set.repository_id, operation_request_id=self._request_id( operation_request_id, "agent-delta", ), expected_repo_state_digest=self.git.repo_state_token( prepared.agent_workspace.agent_worktree ).digest, paths=(prepared.target_path,), path_sources={prepared.relative_path: source}, target_role="agent", target_id=prepared.agent_workspace.record.workspace_id, actor_id=actor_id, trigger=trigger, message=f"Checkpoint {prepared.relative_path}", worktree_root=prepared.agent_workspace.agent_worktree, ) expected_state = item.item_state self.journal.update_git_change_set_item_state( change_set_id=prepared.change_set.change_set_id, relative_path=prepared.relative_path, expected_state=expected_state, state="checkpointed", ) if prepared.overlay is not None: self.journal.update_workspace_overlay_entry_state( snapshot_id=prepared.overlay.snapshot.snapshot_id, relative_path=prepared.relative_path, expected_state="imported_preimage", state="agent_modified", ) self._complete_intent(prepared.intent) if not merge_after_checkpoint: return checkpoint.commit_oid outcome = self.workforce.merge_agent_workspace( prepared.agent_workspace, operation_request_id=self._request_id( operation_request_id, "merge-agent", ), ) return outcome.merged_commit or checkpoint.commit_oid def _prepare_direct_file_write( self, *, context: RunContext, run, repository_root: Path, binding: ProjectWorkspaceBindingRecord, relative_path: str, operation_request_id: str, actor_id: str, trigger: str, ) -> PreparedWorkspaceWrite: root = self._require_direct_checkout( context=context, run=run, repository_root=repository_root, binding=binding, ) run = self._preserve_direct_preimage( context=context, run=run, binding=binding, root=root, operation_request_id=operation_request_id, ) target = root / relative_path preimage_digest = self._digest_file(target) change_set = self._ensure_direct_change_set( context=context, run=run, binding=binding, ) existing_intent = next( ( item for item in self.journal.list_git_mutation_intents() if item.change_set_id == change_set.change_set_id and item.operation_request_id == operation_request_id ), None, ) if existing_intent is not None: preimage_digest = existing_intent.preimage_digest intent = self.journal.ensure_git_mutation_intent( intent_id=self._intent_id( change_set.change_set_id, operation_request_id, ), change_set_id=change_set.change_set_id, operation_request_id=operation_request_id, mutation_scope="exact_path", relative_path=relative_path, preimage_digest=preimage_digest, actor_id=actor_id, trigger=trigger, ) return PreparedWorkspaceWrite( context=context, workspace=None, agent_workspace=None, change_set=change_set, intent=intent, relative_path=relative_path, target_path=target, preimage_digest=preimage_digest, overlay=None, direct_binding=binding, ) def _prepare_direct_broad_write( self, *, context: RunContext, run, repository_root: Path, binding: ProjectWorkspaceBindingRecord, operation_request_id: str, actor_id: str, trigger: str, ) -> PreparedWorkspaceExecution: root = self._require_direct_checkout( context=context, run=run, repository_root=repository_root, binding=binding, ) run = self._preserve_direct_preimage( context=context, run=run, binding=binding, root=root, operation_request_id=operation_request_id, ) change_set = self._ensure_direct_change_set( context=context, run=run, binding=binding, ) intent = self.journal.ensure_git_mutation_intent( intent_id=self._intent_id( change_set.change_set_id, operation_request_id, ), change_set_id=change_set.change_set_id, operation_request_id=operation_request_id, mutation_scope="broad_process", relative_path=None, preimage_digest=None, actor_id=actor_id, trigger=trigger, ) return PreparedWorkspaceExecution( context=context, workspace=None, agent_workspace=None, change_set=change_set, intent=intent, imported_overlays=(), direct_binding=binding, ) def _require_direct_checkout( self, *, context: RunContext, run, repository_root: Path, binding: ProjectWorkspaceBindingRecord, ) -> Path: root = Path(binding.worktree_path).expanduser().resolve() repository_root = repository_root.expanduser().resolve() if binding.checkout_mode == "primary_checkout": if root != repository_root: raise ContentRepositoryError( "primary checkout binding no longer matches the Space root" ) elif not any( item.path == root and item.ref_name == binding.target_ref for item in self.git.list_worktrees(repository_root) ): raise ContentRepositoryError( "explicit Project worktree is no longer registered for its " "selected branch" ) if binding.repository_id != run.repository_id: raise ContentRepositoryError( "Project checkout belongs to another Content Repository" ) probe = self.git.probe(root) observed_ref = f"refs/heads/{probe.branch}" if probe.branch else None if observed_ref != binding.target_ref: raise ContentRepositoryError( "bound checkout is on another branch; switch it explicitly " "before this Task writes" ) request = self.journal.get_workspace_writer_request( f"workspace-writer:{context.run_id}" ) if ( request is None or request.status != "acquired" or request.task_id != context.task_id or request.checkout_id != binding.checkout_id ): raise ContentRepositoryError( "Task does not own the bound checkout writer lease" ) return root def _preserve_direct_preimage( self, *, context: RunContext, run, binding: ProjectWorkspaceBindingRecord, root: Path, operation_request_id: str, ): managed = any( self.journal.list_git_change_set_items(item.change_set_id) for item in self.journal.list_git_change_sets() if item.run_id == context.run_id ) if managed: return run status = self.git.worktree_status(root) if not status: return run paths = tuple(root / path for path in sorted(status)) checkpoint = self.content.checkpoint( run.repository_id, operation_request_id=self._request_id( operation_request_id, "direct-user-preimage", ), expected_repo_state_digest=self.git.repo_state_token(root).digest, paths=paths, path_sources={path: "user_selected" for path in sorted(status)}, target_role="user", target_id=context.project_id, actor_id=context.user_id or "local-user", trigger="workspace.preimage", message=f"Save workspace before Task {context.task_id}", worktree_root=root, commit_trailers={ "Eigent-Initiator": "User", "Eigent-Run-ID": context.run_id, "Eigent-Task-ID": context.task_id, }, ) return self.journal.rebase_unmaterialized_git_run( run_id=context.run_id, expected_base_commit=run.workspace_base_commit, base_commit=checkpoint.commit_oid, base_ref=binding.target_ref, ) def _ensure_direct_change_set( self, *, context: RunContext, run, binding: ProjectWorkspaceBindingRecord, ) -> GitChangeSetRecord: base_commit = run.workspace_base_commit or self.git.current_head( Path(binding.worktree_path) ) if base_commit is None: base_commit = self.git.empty_tree_oid(Path(binding.worktree_path)) run = self.journal.rebase_unmaterialized_git_run( run_id=context.run_id, expected_base_commit=None, base_commit=base_commit, base_ref=binding.target_ref, ) return self.journal.ensure_git_change_set( change_set_id=( "changeset_" + canonical_digest( { "run_id": context.run_id, "checkout_id": binding.checkout_id, "target_ref": binding.target_ref, } )[:32] ), run_id=context.run_id, repository_id=run.repository_id, worktree_ref=binding.target_ref, base_commit=base_commit, ) def _complete_direct_file_write( self, prepared: PreparedWorkspaceWrite, *, operation_request_id: str, actor_id: str, trigger: str, ) -> str | None: result_digest = self._digest_file(prepared.target_path) if result_digest == prepared.preimage_digest: self._complete_intent(prepared.intent) return None size = ( prepared.target_path.stat().st_size if result_digest is not None else None ) tracked = self.git.is_tracked( prepared.mutation_root, prepared.target_path, ) change_kind = ( "deleted" if result_digest is None else ("modified" if tracked else "added") ) source = ( "agent_created" if change_kind == "added" else "agent_modified" ) item = self.journal.put_git_change_set_item( change_set_id=prepared.change_set.change_set_id, relative_path=prepared.relative_path, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, change_kind=change_kind, source=source, preimage_digest=prepared.preimage_digest, result_digest=result_digest, size_bytes=size, ) if item.item_state == "checkpointed": self._complete_intent(prepared.intent) return self.git.current_head(prepared.mutation_root) checkpoint = self.content.checkpoint( prepared.change_set.repository_id, operation_request_id=self._request_id( operation_request_id, "direct-delta", ), expected_repo_state_digest=self.git.repo_state_token( prepared.mutation_root ).digest, paths=(prepared.target_path,), path_sources={prepared.relative_path: source}, target_role="run", target_id=prepared.context.run_id, actor_id=actor_id, trigger=trigger, message=f"Update {prepared.relative_path}", worktree_root=prepared.mutation_root, commit_trailers=self._direct_commit_trailers(prepared.context), ) self.journal.update_git_change_set_item_state( change_set_id=prepared.change_set.change_set_id, relative_path=prepared.relative_path, expected_state=item.item_state, state="checkpointed", ) self._complete_intent(prepared.intent) return checkpoint.commit_oid def _complete_direct_broad_write( self, prepared: PreparedWorkspaceExecution, *, operation_request_id: str, actor_id: str, trigger: str, ) -> tuple[str, ...]: root = prepared.mutation_root status = self.git.worktree_status(root) if not status: self._complete_intent(prepared.intent) return () paths: list[Path] = [] sources: dict[str, str] = {} items = [] for relative_path in sorted(status): target = root / relative_path tracked = self.git.is_tracked(root, target) result_digest = self._digest_file(target) change_kind = ( "deleted" if result_digest is None else ("modified" if tracked else "added") ) source = ( "agent_created" if change_kind == "added" else "agent_modified" ) item = self.journal.put_git_change_set_item( change_set_id=prepared.change_set.change_set_id, relative_path=relative_path, operation_request_id=operation_request_id, actor_id=actor_id, trigger=trigger, change_kind=change_kind, source="worktree_delta", preimage_digest=None, result_digest=result_digest, size_bytes=( target.stat().st_size if target.is_file() else None ), ) paths.append(target) sources[relative_path] = source items.append(item) checkpoint = self.content.checkpoint( prepared.change_set.repository_id, operation_request_id=self._request_id( operation_request_id, "direct-terminal-delta", ), expected_repo_state_digest=self.git.repo_state_token(root).digest, paths=tuple(paths), path_sources=sources, target_role="run", target_id=prepared.context.run_id, actor_id=actor_id, trigger=trigger, message="Checkpoint Task workspace changes", worktree_root=root, commit_trailers=self._direct_commit_trailers(prepared.context), ) for item in items: if item.item_state != "checkpointed": self.journal.update_git_change_set_item_state( change_set_id=prepared.change_set.change_set_id, relative_path=item.relative_path, expected_state=item.item_state, state="checkpointed", ) self._complete_intent(prepared.intent) return (checkpoint.commit_oid,) @staticmethod def _direct_commit_trailers(context: RunContext) -> dict[str, str]: return { "Eigent-Initiator": "Agent", "Eigent-Run-ID": context.run_id, "Eigent-Task-ID": context.task_id, "Eigent-Task-Status": "checkpointed", } def _complete_intent(self, intent: GitMutationIntentRecord) -> None: self.journal.update_git_mutation_intent_status( intent_id=intent.intent_id, expected_status="prepared", status="completed", ) def reconcile_startup(self) -> WorkspaceMutationReconciliation: """Converge interrupted private-worktree checkpoints after restart. Recovery never reads a mutable User source as an authoritative preimage and never writes into the User Worktree. One ambiguous ChangeSet is quarantined without blocking reconciliation of others. """ self.journal.reset_git_agent_workspace_leases_after_restart() recovered: list[str] = [] needs_attention: list[str] = [] change_sets = { item.change_set_id: item for item in self.journal.list_git_change_sets() } for intent in self.journal.list_git_mutation_intents( statuses=("prepared",) ): change_set = change_sets.get(intent.change_set_id) if change_set is None or change_set.state != "open": continue try: self._reconcile_intent(change_set, intent) except Exception: logger.exception( "Workspace mutation intent startup reconciliation needs attention", extra={"intent_id": intent.intent_id}, ) self.journal.update_git_mutation_intent_status( intent_id=intent.intent_id, expected_status="prepared", status="needs_attention", ) self.journal.update_git_change_set_state( change_set_id=change_set.change_set_id, expected_state="open", state="needs_attention", ) needs_attention.append(change_set.change_set_id) continue if change_set.change_set_id not in recovered: recovered.append(change_set.change_set_id) for change_set in self.journal.list_git_change_sets(states=("open",)): items = self.journal.list_git_change_set_items( change_set.change_set_id, states=("pending", "preimage_checkpointed"), ) if not items: continue try: self._reconcile_change_set(change_set, items) except Exception: logger.exception( "Workspace ChangeSet startup reconciliation needs attention", extra={"change_set_id": change_set.change_set_id}, ) self.journal.update_git_change_set_state( change_set_id=change_set.change_set_id, expected_state="open", state="needs_attention", ) needs_attention.append(change_set.change_set_id) continue if change_set.change_set_id not in recovered: recovered.append(change_set.change_set_id) for change_set in self.journal.list_git_change_sets(states=("open",)): checkpointed = self.journal.list_git_change_set_items( change_set.change_set_id, states=("checkpointed",), ) if not checkpointed: continue agent_record = next( ( item for item in self.journal.list_git_agent_workspaces( run_id=change_set.run_id, states=("ready",), ) if item.agent_ref == change_set.worktree_ref ), None, ) if agent_record is None: continue workspace = self._workspace_for_change_set(change_set) agent_head = self.git.current_head( Path(agent_record.worktree_path) ) run_head = self.git.current_head(workspace.run_worktree) if ( agent_head is None or run_head is None or agent_head == run_head ): continue try: agent_workspace = self.workforce.ensure_agent_workspace( run_workspace=workspace, agent_id=agent_record.agent_id, operation_request_id=( f"startup-checkpointed-merge:{change_set.change_set_id}" ), ) self.workforce.merge_agent_workspace( agent_workspace, operation_request_id=self._request_id( change_set.change_set_id, "merge-agent-recovery", ), ) except Exception: logger.exception( "Checkpointed Agent result merge needs attention", extra={"change_set_id": change_set.change_set_id}, ) self.journal.update_git_change_set_state( change_set_id=change_set.change_set_id, expected_state="open", state="needs_attention", ) needs_attention.append(change_set.change_set_id) continue if change_set.change_set_id not in recovered: recovered.append(change_set.change_set_id) return WorkspaceMutationReconciliation( recovered_change_set_ids=tuple(recovered), needs_attention_change_set_ids=tuple(needs_attention), ) def _reconcile_intent(self, change_set, intent) -> None: workspace = self._workspace_for_change_set(change_set) agent_workspace = self.workforce.ensure_agent_workspace( run_workspace=workspace, agent_id=intent.actor_id, operation_request_id=intent.operation_request_id, ) if intent.mutation_scope == "exact_path": if intent.relative_path is None: raise ContentRepositoryError( "Exact-path mutation intent has no path" ) target = agent_workspace.agent_worktree / intent.relative_path overlay = self._materialized_overlay_for_path( change_set.run_id, intent.relative_path, target, ) persisted_item = next( ( item for item in self.journal.list_git_change_set_items( change_set.change_set_id, states=("pending", "preimage_checkpointed"), ) if item.relative_path == intent.relative_path ), None, ) if persisted_item is not None: self._reconcile_exact_item( change_set, workspace, agent_workspace, persisted_item, overlay, ) self._complete_intent(intent) self.workforce.merge_agent_workspace( agent_workspace, operation_request_id=self._request_id( intent.operation_request_id, "merge-agent-recovery", ), ) return self.complete_file_write( PreparedWorkspaceWrite( context=self._reconciliation_context(change_set), workspace=workspace, agent_workspace=agent_workspace, change_set=change_set, intent=intent, relative_path=intent.relative_path, target_path=target, preimage_digest=intent.preimage_digest, overlay=overlay, ), operation_request_id=intent.operation_request_id, actor_id=intent.actor_id, trigger=intent.trigger, ) return if intent.mutation_scope != "broad_process": raise ContentRepositoryError( f"Unsupported mutation scope {intent.mutation_scope!r}" ) imported = self._materialized_overlays( change_set.run_id, agent_workspace.agent_worktree, ) self.complete_broad_write( PreparedWorkspaceExecution( context=self._reconciliation_context(change_set), workspace=workspace, agent_workspace=agent_workspace, change_set=change_set, intent=intent, imported_overlays=imported, ), operation_request_id=intent.operation_request_id, actor_id=intent.actor_id, trigger=intent.trigger, ) def _workspace_for_change_set(self, change_set) -> GitRunWorkspace: run = self.journal.get_run_git_materialization(change_set.run_id) if run is None or not run.worktree_path or not run.run_ref: raise ContentRepositoryError( "ChangeSet Run worktree is unavailable during reconciliation" ) project = self.journal.get_project_git_state(run.project_id) if project is None or not project.worktree_path: raise ContentRepositoryError( "ChangeSet Project worktree is unavailable during reconciliation" ) return GitRunWorkspace( project=project, run=run, project_worktree=Path(project.worktree_path), run_worktree=Path(run.worktree_path), ) def _materialized_overlay_for_path( self, run_id: str, relative_path: str, destination_path: Path, ) -> MaterializedOverlay | None: snapshot = self.snapshots.get_snapshot(run_id) if snapshot is None: return None entry = self.journal.get_workspace_overlay_entry( snapshot.snapshot_id, relative_path, ) if ( entry is None or entry.source_kind != "user_overlay" or entry.entry_state == "agent_modified" ): return None if ( entry.materialized_content_digest is None or entry.preimage_cache_key is None ): raise ContentRepositoryError( "Overlay recovery metadata is incomplete" ) return MaterializedOverlay( snapshot=snapshot, entry=entry, relative_path=relative_path, content_digest=entry.materialized_content_digest, size_bytes=entry.size_bytes, preimage_path=( self.snapshots.cache_root / "preimages" / entry.preimage_cache_key[:2] / entry.preimage_cache_key ), destination_path=destination_path, ) def _materialized_overlays( self, run_id: str, worktree_root: Path, ) -> tuple[MaterializedOverlay, ...]: snapshot = self.snapshots.get_snapshot(run_id) if snapshot is None: return () values: list[MaterializedOverlay] = [] for entry in self.journal.list_workspace_overlay_entries( snapshot.snapshot_id ): overlay = self._materialized_overlay_for_path( run_id, entry.relative_path, worktree_root / entry.relative_path, ) if overlay is not None: values.append(overlay) return tuple(values) def _reconciliation_context(self, change_set) -> RunContext: repository = self.journal.get_git_repository(change_set.repository_id) if repository is None: raise ContentRepositoryError( "ChangeSet repository is unavailable during reconciliation" ) run = self.journal.get_run(change_set.run_id) if run is None: raise ContentRepositoryError( "ChangeSet Run is unavailable during reconciliation" ) root = Path(repository.root_path) return RunContext( space_id=repository.space_id, project_id=run.project_id, run_id=run.run_id, task_id=run.run_id, email="startup-reconciliation", user_id="startup-reconciliation", working_directory=root, task_output_root=root, camel_log_dir=root / ".eigent" / "logs", binding_source="startup-reconciliation", workdir_mode="direct-write", browser_port=0, ) def _reconcile_change_set(self, change_set, items) -> None: workspace = self._workspace_for_change_set(change_set) agent_workspace = self.workforce.ensure_agent_workspace( run_workspace=workspace, agent_id=items[0].actor_id, operation_request_id=items[0].operation_request_id, ) snapshot = self.snapshots.get_snapshot(change_set.run_id) overlay_by_path = {} if snapshot is not None: overlay_by_path = { entry.relative_path: entry for entry in self.journal.list_workspace_overlay_entries( snapshot.snapshot_id ) if entry.source_kind == "user_overlay" } broad = [item for item in items if item.source == "worktree_delta"] exact = [item for item in items if item.source != "worktree_delta"] if broad: self._reconcile_broad_items( change_set, workspace, agent_workspace, broad, ) for item in exact: overlay_entry = overlay_by_path.get(item.relative_path) overlay = None if overlay_entry is not None: if ( snapshot is None or overlay_entry.materialized_content_digest is None or overlay_entry.preimage_cache_key is None ): raise ContentRepositoryError( "Overlay recovery metadata is incomplete" ) overlay = MaterializedOverlay( snapshot=snapshot, entry=overlay_entry, relative_path=item.relative_path, content_digest=overlay_entry.materialized_content_digest, size_bytes=overlay_entry.size_bytes, preimage_path=( self.snapshots.cache_root / "preimages" / overlay_entry.preimage_cache_key[:2] / overlay_entry.preimage_cache_key ), destination_path=agent_workspace.agent_worktree / item.relative_path, ) self._reconcile_exact_item( change_set, workspace, agent_workspace, item, overlay, ) self.workforce.merge_agent_workspace( agent_workspace, operation_request_id=self._request_id( items[0].operation_request_id, "merge-agent-recovery", ), ) def _reconcile_broad_items( self, change_set, workspace, agent_workspace, items, ) -> None: request_ids = {item.operation_request_id for item in items} if len(request_ids) != 1: raise ContentRepositoryError( "Broad ChangeSet recovery has mixed operation requests" ) paths: list[Path] = [] sources: dict[str, str] = {} for item in items: target = agent_workspace.agent_worktree / item.relative_path if self._digest_file(target) != item.result_digest: raise ContentRepositoryError( f"Run result for {item.relative_path!r} changed before recovery" ) paths.append(target) sources[item.relative_path] = ( "agent_created" if item.change_kind == "added" else "agent_modified" ) checkpoint_request_id = self._request_id( next(iter(request_ids)), "terminal-delta", ) self.content.checkpoint( change_set.repository_id, operation_request_id=checkpoint_request_id, expected_repo_state_digest=self._checkpoint_expected_digest( repository_id=change_set.repository_id, operation_request_id=checkpoint_request_id, worktree_root=agent_workspace.agent_worktree, ), paths=tuple(paths), path_sources=sources, target_role="agent", target_id=agent_workspace.record.workspace_id, actor_id=items[0].actor_id, trigger=items[0].trigger, message="Checkpoint bounded workspace process delta", worktree_root=agent_workspace.agent_worktree, ) for item in items: self.journal.update_git_change_set_item_state( change_set_id=change_set.change_set_id, relative_path=item.relative_path, expected_state=item.item_state, state="checkpointed", ) def _reconcile_exact_item( self, change_set, workspace, agent_workspace, item, overlay: MaterializedOverlay | None, ) -> None: target = agent_workspace.agent_worktree / item.relative_path if overlay is not None: result_cache = self._result_cache_path(item.result_digest) current_digest = self._digest_file(target) if item.result_digest is not None and not result_cache.exists(): if current_digest != item.result_digest: raise ContentRepositoryError( f"Agent result cache for {item.relative_path!r} is unavailable" ) self._cache_result( target, expected_digest=item.result_digest, ) if item.item_state == "pending": if not overlay.preimage_path.is_file(): raise ContentRepositoryError( f"Overlay preimage for {item.relative_path!r} is unavailable" ) self._replace_from_file(overlay.preimage_path, target) checkpoint_request_id = self._request_id( item.operation_request_id, "overlay-preimage", ) self.content.checkpoint( change_set.repository_id, operation_request_id=checkpoint_request_id, expected_repo_state_digest=self._checkpoint_expected_digest( repository_id=change_set.repository_id, operation_request_id=checkpoint_request_id, worktree_root=agent_workspace.agent_worktree, ), paths=(target,), path_sources={item.relative_path: "overlay_preimage"}, target_role="agent", target_id=agent_workspace.record.workspace_id, actor_id="user", trigger="overlay_preimage", message=f"Preserve User preimage for {item.relative_path}", worktree_root=agent_workspace.agent_worktree, ) self.journal.update_git_change_set_item_state( change_set_id=change_set.change_set_id, relative_path=item.relative_path, expected_state="pending", state="preimage_checkpointed", ) if item.result_digest is None: target.unlink(missing_ok=True) else: self._replace_from_file(result_cache, target) elif self._digest_file(target) != item.result_digest: raise ContentRepositoryError( f"Run result for {item.relative_path!r} changed before recovery" ) source = ( "agent_created" if item.change_kind == "added" else "agent_modified" ) checkpoint_request_id = self._request_id( item.operation_request_id, "agent-delta", ) self.content.checkpoint( change_set.repository_id, operation_request_id=checkpoint_request_id, expected_repo_state_digest=self._checkpoint_expected_digest( repository_id=change_set.repository_id, operation_request_id=checkpoint_request_id, worktree_root=agent_workspace.agent_worktree, ), paths=(target,), path_sources={item.relative_path: source}, target_role="agent", target_id=agent_workspace.record.workspace_id, actor_id=item.actor_id, trigger=item.trigger, message=f"Checkpoint {item.relative_path}", worktree_root=agent_workspace.agent_worktree, ) self.journal.update_git_change_set_item_state( change_set_id=change_set.change_set_id, relative_path=item.relative_path, expected_state=( "preimage_checkpointed" if overlay is not None else item.item_state ), state="checkpointed", ) if ( overlay is not None and overlay.entry.entry_state == "imported_preimage" ): self.journal.update_workspace_overlay_entry_state( snapshot_id=overlay.snapshot.snapshot_id, relative_path=item.relative_path, expected_state="imported_preimage", state="agent_modified", ) def _result_cache_path(self, digest: str | None) -> Path: if digest is None: return self.state_root / "snapshots" / "results" / "deleted" return self.state_root / "snapshots" / "results" / digest[:2] / digest def _checkpoint_expected_digest( self, *, repository_id: str, operation_request_id: str, worktree_root: Path, ) -> str: operation_id = ( "gitop_" + canonical_digest( { "repository_id": repository_id, "request_id": operation_request_id, } )[:32] ) operation = self.journal.get_git_operation(operation_id) if ( operation is not None and operation.expected_repo_state_digest is not None ): return operation.expected_repo_state_digest return self.git.repo_state_token(worktree_root).digest def _cache_result(self, path: Path, *, expected_digest: str) -> Path: target = ( self.state_root / "snapshots" / "results" / expected_digest[:2] / expected_digest ) target.parent.mkdir(parents=True, exist_ok=True) if target.exists(): if self._digest_file(target) != expected_digest: raise ContentRepositoryError("result cache digest collision") return target temporary = target.with_name(f".{target.name}.{uuid.uuid4().hex}.tmp") try: self._replace_from_file(path, temporary) if self._digest_file(temporary) != expected_digest: raise ContentRepositoryError("result changed during caching") temporary.chmod(0o600) os.replace(temporary, target) finally: temporary.unlink(missing_ok=True) return target @staticmethod def _replace_from_file(source: Path, destination: Path) -> None: destination.parent.mkdir(parents=True, exist_ok=True) temporary = destination.with_name( f".{destination.name}.{uuid.uuid4().hex}.tmp" ) try: with ( source.open("rb", buffering=0) as reader, temporary.open("wb", buffering=0) as writer, ): while chunk := reader.read(1024 * 1024): writer.write(chunk) os.replace(temporary, destination) finally: temporary.unlink(missing_ok=True) @staticmethod def _digest_file(path: Path) -> str | None: try: if path.is_symlink() or not path.is_file(): return None digest = hashlib.sha256() with path.open("rb", buffering=0) as stream: while chunk := stream.read(1024 * 1024): digest.update(chunk) return digest.hexdigest() except FileNotFoundError: return None @staticmethod def _relative_workspace_path( *, context: RunContext, repository_root: Path, filename: str, ) -> str: root = repository_root.expanduser().resolve() candidate = Path(filename).expanduser() if not candidate.is_absolute(): candidate = context.working_directory / candidate resolved = candidate.resolve() try: relative = resolved.relative_to(root) except ValueError as exc: raise ContentRepositoryError( "Workspace write target is outside the Content Repository" ) from exc if not relative.parts: raise ContentRepositoryError( "Workspace write target is not a file" ) return relative.as_posix() @staticmethod def _request_id(value: str, suffix: str) -> str: return ( "workspace-mutation:" + canonical_digest({"request_id": value, "phase": suffix})[:48] ) @staticmethod def _intent_id(change_set_id: str, operation_request_id: str) -> str: return ( "gitintent_" + canonical_digest( { "change_set_id": change_set_id, "operation_request_id": operation_request_id, } )[:32] ) def get_default_workspace_mutation_service() -> WorkspaceMutationService: return WorkspaceMutationService( get_default_run_journal(), state_root=configured_run_journal_path().parent / "workspace-git", )