mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-27 17:41:56 +00:00
1130 lines
41 KiB
Python
1130 lines
41 KiB
Python
# ========= 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. =========
|
|
|
|
"""Per-agent worktrees and serialized integration into a Run workspace."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from app.run_journal import (
|
|
GitAgentWorkspaceRecord,
|
|
RunEventDraft,
|
|
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,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GitAgentWorkspace:
|
|
record: GitAgentWorkspaceRecord
|
|
run_workspace: GitRunWorkspace
|
|
agent_worktree: Path
|
|
lease_token: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GitAgentMergeOutcome:
|
|
workspace: GitAgentWorkspaceRecord
|
|
status: str
|
|
merged_commit: str | None
|
|
conflict_paths: tuple[str, ...]
|
|
interaction_id: str | None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class GitAgentReconciliation:
|
|
recovered_workspace_ids: tuple[str, ...]
|
|
needs_attention_workspace_ids: tuple[str, ...]
|
|
|
|
|
|
class WorkforceGitService:
|
|
"""Own one writable worktree per Run/Agent and merge under one lock."""
|
|
|
|
def __init__(
|
|
self,
|
|
journal: SQLiteRunJournal,
|
|
*,
|
|
state_root: Path,
|
|
coordinator: WorkspaceGitCoordinator | None = None,
|
|
lease_seconds: float = 300.0,
|
|
) -> None:
|
|
if lease_seconds <= 0:
|
|
raise ValueError("Agent workspace lease must be positive")
|
|
self.journal = journal
|
|
self.state_root = state_root.expanduser().resolve()
|
|
self.coordinator = coordinator or WorkspaceGitCoordinator(
|
|
journal,
|
|
state_root=self.state_root,
|
|
)
|
|
self.git = self.coordinator.git
|
|
self.content = self.coordinator.content
|
|
self.lease_seconds = lease_seconds
|
|
|
|
def ensure_agent_workspace(
|
|
self,
|
|
*,
|
|
run_workspace: GitRunWorkspace,
|
|
agent_id: str,
|
|
operation_request_id: str,
|
|
now: float | None = None,
|
|
) -> GitAgentWorkspace:
|
|
timestamp = now if now is not None else time.time()
|
|
if not agent_id.strip() or not operation_request_id.strip():
|
|
raise ValueError("Agent id and operation request id are required")
|
|
run = run_workspace.run
|
|
repository = self.journal.get_git_repository(run.repository_id)
|
|
if repository is None or run.run_ref is None:
|
|
raise ContentRepositoryError(
|
|
"Run Agent workspace owner is missing"
|
|
)
|
|
run_head = self.git.current_head(run_workspace.run_worktree)
|
|
if run_head is None:
|
|
raise ContentRepositoryError("Run workspace has no HEAD")
|
|
existing = self.journal.get_git_agent_workspace(run.run_id, agent_id)
|
|
if existing is not None and existing.state in {
|
|
"conflicted",
|
|
"needs_attention",
|
|
"archived",
|
|
}:
|
|
raise ContentRepositoryError(
|
|
f"Agent workspace is {existing.state!r}"
|
|
)
|
|
base_commit = (
|
|
existing.base_commit if existing is not None else run_head
|
|
)
|
|
workspace_id = (
|
|
"agentws_"
|
|
+ canonical_digest({"run_id": run.run_id, "agent_id": agent_id})[
|
|
:32
|
|
]
|
|
)
|
|
agent_ref = (
|
|
existing.agent_ref
|
|
if existing is not None
|
|
else self._agent_ref(run.repository_id, run.run_id, agent_id)
|
|
)
|
|
agent_worktree = (
|
|
Path(existing.worktree_path)
|
|
if existing is not None
|
|
else self._agent_worktree(
|
|
repository.space_id, run.run_id, agent_id
|
|
)
|
|
)
|
|
lease_token = canonical_digest(
|
|
{
|
|
"workspace_id": workspace_id,
|
|
"operation_request_id": operation_request_id,
|
|
}
|
|
)
|
|
record = self.journal.claim_git_agent_workspace(
|
|
workspace_id=workspace_id,
|
|
run_id=run.run_id,
|
|
repository_id=run.repository_id,
|
|
agent_id=agent_id,
|
|
agent_ref=agent_ref,
|
|
worktree_path=str(agent_worktree),
|
|
base_commit=base_commit,
|
|
lease_owner=operation_request_id,
|
|
lease_token=lease_token,
|
|
lease_until=timestamp + self.lease_seconds,
|
|
now=timestamp,
|
|
)
|
|
|
|
if record.state in {"admitted", "materializing"}:
|
|
record = self._materialize(
|
|
record,
|
|
run_workspace=run_workspace,
|
|
lease_token=lease_token,
|
|
operation_request_id=operation_request_id,
|
|
now=timestamp,
|
|
)
|
|
elif record.state == "merging":
|
|
record = self._recover_dispatched_merge(
|
|
record,
|
|
run_workspace=run_workspace,
|
|
lease_token=lease_token,
|
|
now=timestamp,
|
|
)
|
|
if record.state in {"ready", "merged"}:
|
|
record = self._refresh_agent_projection(
|
|
record,
|
|
run_workspace=run_workspace,
|
|
lease_token=lease_token,
|
|
run_head=run_head,
|
|
now=timestamp,
|
|
)
|
|
if record.state != "ready":
|
|
raise ContentRepositoryError(
|
|
f"Agent workspace is {record.state!r}"
|
|
)
|
|
return GitAgentWorkspace(
|
|
record=record,
|
|
run_workspace=run_workspace,
|
|
agent_worktree=agent_worktree,
|
|
lease_token=lease_token,
|
|
)
|
|
|
|
def merge_agent_workspace(
|
|
self,
|
|
workspace: GitAgentWorkspace,
|
|
*,
|
|
operation_request_id: str,
|
|
now: float | None = None,
|
|
) -> GitAgentMergeOutcome:
|
|
timestamp = now if now is not None else time.time()
|
|
record = self.journal.get_git_agent_workspace(
|
|
workspace.record.run_id, workspace.record.agent_id
|
|
)
|
|
if record is None or record.lease_token != workspace.lease_token:
|
|
raise ContentRepositoryError(
|
|
"Agent workspace lease is no longer held"
|
|
)
|
|
if record.state not in {"ready", "merging"}:
|
|
raise ContentRepositoryError(
|
|
f"Agent workspace cannot merge from {record.state!r}"
|
|
)
|
|
if not self.git.is_worktree_clean(workspace.agent_worktree):
|
|
raise ContentRepositoryError(
|
|
"Agent worktree has uncheckpointed changes"
|
|
)
|
|
repository = self.journal.get_git_repository(record.repository_id)
|
|
if repository is None:
|
|
raise ContentRepositoryError("Agent repository is unavailable")
|
|
operation_id = (
|
|
"gitop_"
|
|
+ canonical_digest(
|
|
{
|
|
"repository_id": record.repository_id,
|
|
"request_id": operation_request_id,
|
|
}
|
|
)[:32]
|
|
)
|
|
payload = {
|
|
"workspace_id": record.workspace_id,
|
|
"run_id": record.run_id,
|
|
"agent_id": record.agent_id,
|
|
"agent_ref": record.agent_ref,
|
|
"agent_head": self.git.current_head(workspace.agent_worktree),
|
|
}
|
|
run_root = workspace.run_workspace.run_worktree
|
|
with self.content.repository_lock(
|
|
self.content.repository_lock_path(repository.space_id)
|
|
):
|
|
current_run_state = self.git.repo_state_token(run_root)
|
|
existing_operation = self.journal.get_git_operation(operation_id)
|
|
operation = self.journal.begin_git_operation(
|
|
operation_id=operation_id,
|
|
repository_id=record.repository_id,
|
|
request_id=operation_request_id,
|
|
operation_type="agent.merge",
|
|
payload_digest=canonical_digest(payload),
|
|
expected_repo_state_digest=(
|
|
existing_operation.expected_repo_state_digest
|
|
if existing_operation is not None
|
|
else current_run_state.digest
|
|
),
|
|
)
|
|
if record.state == "ready":
|
|
record = self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=workspace.lease_token,
|
|
expected_state="ready",
|
|
state="merging",
|
|
last_operation_id=operation_id,
|
|
now=timestamp,
|
|
)
|
|
if operation.status == "prepared":
|
|
self.journal.mark_git_operation_dispatched(
|
|
operation_id,
|
|
observed_repo_state_digest=current_run_state.digest,
|
|
)
|
|
recovered = self.git.find_commit_by_operation(
|
|
run_root, operation_id
|
|
)
|
|
if recovered is not None:
|
|
return self._complete_merge(
|
|
record,
|
|
workspace=workspace,
|
|
operation_id=operation_id,
|
|
merged_commit=recovered,
|
|
now=timestamp,
|
|
)
|
|
result = self.git.merge_owned_ref(
|
|
run_root,
|
|
source_ref=record.agent_ref,
|
|
operation_id=operation_id,
|
|
message=f"Merge Agent {record.agent_id} into Run {record.run_id}",
|
|
)
|
|
if result.conflict_paths:
|
|
return self._record_conflict(
|
|
record,
|
|
workspace=workspace,
|
|
operation_id=operation_id,
|
|
conflict_paths=result.conflict_paths,
|
|
now=timestamp,
|
|
)
|
|
assert result.commit_oid is not None
|
|
return self._complete_merge(
|
|
record,
|
|
workspace=workspace,
|
|
operation_id=operation_id,
|
|
merged_commit=result.commit_oid,
|
|
now=timestamp,
|
|
)
|
|
|
|
def release_workspace(self, workspace: GitAgentWorkspace) -> None:
|
|
self.journal.release_git_agent_workspace_lease(
|
|
workspace.record.workspace_id,
|
|
lease_token=workspace.lease_token,
|
|
)
|
|
|
|
def renew_workspace(
|
|
self,
|
|
workspace: GitAgentWorkspace,
|
|
*,
|
|
now: float | None = None,
|
|
) -> GitAgentWorkspace:
|
|
"""Keep a long-running Agent operation's workspace lease alive."""
|
|
|
|
timestamp = now if now is not None else time.time()
|
|
record = self.journal.renew_git_agent_workspace_lease(
|
|
workspace.record.workspace_id,
|
|
lease_token=workspace.lease_token,
|
|
lease_until=timestamp + self.lease_seconds,
|
|
now=timestamp,
|
|
)
|
|
return GitAgentWorkspace(
|
|
record=record,
|
|
run_workspace=workspace.run_workspace,
|
|
agent_worktree=workspace.agent_worktree,
|
|
lease_token=workspace.lease_token,
|
|
)
|
|
|
|
def reconcile_startup(
|
|
self, *, now: float | None = None
|
|
) -> GitAgentReconciliation:
|
|
timestamp = now if now is not None else time.time()
|
|
recovered: list[str] = []
|
|
attention: list[str] = []
|
|
self.journal.reset_git_agent_workspace_leases_after_restart(
|
|
now=timestamp
|
|
)
|
|
for record in self.journal.list_git_agent_workspaces(
|
|
states=("materializing", "merging")
|
|
):
|
|
run = self.journal.get_run_git_materialization(record.run_id)
|
|
project = (
|
|
self.journal.get_project_git_state(run.project_id)
|
|
if run is not None
|
|
else None
|
|
)
|
|
if (
|
|
run is None
|
|
or project is None
|
|
or run.worktree_path is None
|
|
or project.worktree_path is None
|
|
):
|
|
attention.append(record.workspace_id)
|
|
continue
|
|
run_workspace = GitRunWorkspace(
|
|
project=project,
|
|
run=run,
|
|
project_worktree=Path(project.worktree_path),
|
|
run_worktree=Path(run.worktree_path),
|
|
)
|
|
try:
|
|
recovered_workspace = self.ensure_agent_workspace(
|
|
run_workspace=run_workspace,
|
|
agent_id=record.agent_id,
|
|
operation_request_id=(
|
|
f"startup-agent-reconcile:{record.workspace_id}"
|
|
),
|
|
now=timestamp,
|
|
)
|
|
self.release_workspace(recovered_workspace)
|
|
except Exception:
|
|
current = self.journal.get_git_agent_workspace(
|
|
record.run_id, record.agent_id
|
|
)
|
|
if current is not None and current.state == "conflicted":
|
|
recovered.append(record.workspace_id)
|
|
else:
|
|
logger.exception(
|
|
"Agent workspace startup reconciliation needs attention",
|
|
extra={"workspace_id": record.workspace_id},
|
|
)
|
|
attention.append(record.workspace_id)
|
|
else:
|
|
if record.workspace_id not in recovered:
|
|
recovered.append(record.workspace_id)
|
|
for record in self.journal.list_git_agent_workspaces(
|
|
states=("conflicted",)
|
|
):
|
|
interaction_id = record.conflict_interaction_id
|
|
interaction = (
|
|
self.journal.get_human_interaction(interaction_id)
|
|
if interaction_id
|
|
else None
|
|
)
|
|
if interaction is None or interaction.status != "resolved":
|
|
continue
|
|
try:
|
|
self.resolve_merge_conflict(interaction.interaction_id)
|
|
except Exception:
|
|
logger.exception(
|
|
"Resolved Agent merge conflict needs attention",
|
|
extra={"workspace_id": record.workspace_id},
|
|
)
|
|
attention.append(record.workspace_id)
|
|
else:
|
|
if record.workspace_id not in recovered:
|
|
recovered.append(record.workspace_id)
|
|
return GitAgentReconciliation(tuple(recovered), tuple(attention))
|
|
|
|
def resolve_merge_conflict(
|
|
self,
|
|
interaction_id: str,
|
|
*,
|
|
now: float | None = None,
|
|
) -> GitAgentMergeOutcome:
|
|
"""Apply one durable merge-conflict decision idempotently."""
|
|
|
|
timestamp = now if now is not None else time.time()
|
|
interaction = self.journal.get_human_interaction(interaction_id)
|
|
if (
|
|
interaction is None
|
|
or interaction.interaction_type != "merge_conflict"
|
|
or interaction.status != "resolved"
|
|
):
|
|
raise ContentRepositoryError(
|
|
"Merge conflict has no resolved HumanInteraction"
|
|
)
|
|
decisions = self.journal.list_human_interaction_decisions(
|
|
interaction_id
|
|
)
|
|
if not decisions:
|
|
raise ContentRepositoryError("Merge conflict decision is missing")
|
|
decision_record = decisions[-1]
|
|
decision = self._conflict_decision(decision_record.decision)
|
|
record = next(
|
|
(
|
|
item
|
|
for item in self.journal.list_git_agent_workspaces(
|
|
run_id=interaction.run_id
|
|
)
|
|
if item.conflict_interaction_id == interaction_id
|
|
),
|
|
None,
|
|
)
|
|
if record is None:
|
|
raise ContentRepositoryError(
|
|
"Merge conflict Agent workspace is unavailable"
|
|
)
|
|
operation_id = (
|
|
"gitop_"
|
|
+ canonical_digest(
|
|
{
|
|
"repository_id": record.repository_id,
|
|
"request_id": (
|
|
f"merge-conflict:{decision_record.decision_id}"
|
|
),
|
|
}
|
|
)[:32]
|
|
)
|
|
if record.state == "merged":
|
|
operation = self.journal.get_git_operation(operation_id)
|
|
run_state = self.journal.get_run_git_materialization(record.run_id)
|
|
if (
|
|
operation is not None
|
|
and operation.status == "dispatched"
|
|
and run_state is not None
|
|
and run_state.worktree_path
|
|
and record.head_commit is not None
|
|
):
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={
|
|
"workspace_id": record.workspace_id,
|
|
"decision": decision,
|
|
"commit_oid": record.head_commit,
|
|
},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
Path(run_state.worktree_path)
|
|
).digest,
|
|
now=timestamp,
|
|
)
|
|
return GitAgentMergeOutcome(
|
|
record,
|
|
"merged",
|
|
record.head_commit,
|
|
(),
|
|
interaction_id,
|
|
)
|
|
if record.state == "needs_attention" and decision == "manual":
|
|
return GitAgentMergeOutcome(
|
|
record,
|
|
"needs_attention",
|
|
None,
|
|
tuple(interaction.request.get("conflict_paths") or ()),
|
|
interaction_id,
|
|
)
|
|
run = self.journal.get_run_git_materialization(record.run_id)
|
|
project = (
|
|
self.journal.get_project_git_state(run.project_id)
|
|
if run is not None
|
|
else None
|
|
)
|
|
repository = self.journal.get_git_repository(record.repository_id)
|
|
if (
|
|
run is None
|
|
or project is None
|
|
or repository is None
|
|
or not run.worktree_path
|
|
or not project.worktree_path
|
|
):
|
|
raise ContentRepositoryError(
|
|
"Merge conflict Run workspace is unavailable"
|
|
)
|
|
run_workspace = GitRunWorkspace(
|
|
project=project,
|
|
run=run,
|
|
project_worktree=Path(project.worktree_path),
|
|
run_worktree=Path(run.worktree_path),
|
|
)
|
|
lease_owner = f"resolve:{interaction_id}"
|
|
lease_token = canonical_digest(
|
|
{
|
|
"workspace_id": record.workspace_id,
|
|
"decision_id": decision_record.decision_id,
|
|
}
|
|
)
|
|
record = self.journal.claim_git_agent_workspace(
|
|
workspace_id=record.workspace_id,
|
|
run_id=record.run_id,
|
|
repository_id=record.repository_id,
|
|
agent_id=record.agent_id,
|
|
agent_ref=record.agent_ref,
|
|
worktree_path=record.worktree_path,
|
|
base_commit=record.base_commit,
|
|
lease_owner=lease_owner,
|
|
lease_token=lease_token,
|
|
lease_until=timestamp + self.lease_seconds,
|
|
now=timestamp,
|
|
)
|
|
conflict_paths = tuple(
|
|
str(path) for path in interaction.request.get("conflict_paths", [])
|
|
)
|
|
if decision == "manual":
|
|
record = self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="conflicted",
|
|
state="needs_attention",
|
|
conflict_interaction_id=interaction_id,
|
|
release_lease=True,
|
|
run_event=self._conflict_resolved_event(
|
|
record,
|
|
interaction_id=interaction_id,
|
|
decision=decision,
|
|
commit_oid=None,
|
|
now=timestamp,
|
|
),
|
|
now=timestamp,
|
|
)
|
|
return GitAgentMergeOutcome(
|
|
record,
|
|
"needs_attention",
|
|
None,
|
|
conflict_paths,
|
|
interaction_id,
|
|
)
|
|
|
|
run_root = run_workspace.run_worktree
|
|
agent_root = Path(record.worktree_path)
|
|
with self.content.repository_lock(
|
|
self.content.repository_lock_path(repository.space_id)
|
|
):
|
|
operation = self.journal.begin_git_operation(
|
|
operation_id=operation_id,
|
|
repository_id=record.repository_id,
|
|
request_id=f"merge-conflict:{decision_record.decision_id}",
|
|
operation_type="agent.resolve_conflict",
|
|
payload_digest=canonical_digest(
|
|
{
|
|
"workspace_id": record.workspace_id,
|
|
"interaction_id": interaction_id,
|
|
"decision": decision,
|
|
"paths": conflict_paths,
|
|
}
|
|
),
|
|
expected_repo_state_digest=self.git.repo_state_token(
|
|
run_root
|
|
).digest,
|
|
)
|
|
if operation.status == "prepared":
|
|
self.journal.mark_git_operation_dispatched(
|
|
operation_id,
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
run_root
|
|
).digest,
|
|
now=timestamp,
|
|
)
|
|
recovered = self.git.find_commit_by_operation(
|
|
run_root, operation_id
|
|
)
|
|
if decision == "take_agent" and recovered is None:
|
|
self.git.restore_owned_paths_from_ref(
|
|
run_root,
|
|
source_ref=record.agent_ref,
|
|
relative_paths=conflict_paths,
|
|
)
|
|
recovered = self.git.commit_paths(
|
|
run_root,
|
|
tuple(run_root / path for path in conflict_paths),
|
|
message=(
|
|
"Resolve Agent merge conflict\n\n"
|
|
f"Eigent-Operation: {operation_id}\n"
|
|
f"Eigent-Interaction: {interaction_id}"
|
|
),
|
|
author_name="Eigent User",
|
|
author_email="noreply@eigent.ai",
|
|
)
|
|
if recovered is None:
|
|
recovered = self.git.current_head(run_root)
|
|
if recovered is None:
|
|
raise ContentRepositoryError("Run workspace has no HEAD")
|
|
current_agent_head = self.git.current_head(agent_root)
|
|
if current_agent_head is None:
|
|
raise ContentRepositoryError("Agent workspace has no HEAD")
|
|
self.git.refresh_owned_worktree(
|
|
agent_root,
|
|
expected_projected_head=current_agent_head,
|
|
target_head=recovered,
|
|
)
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={
|
|
"workspace_id": record.workspace_id,
|
|
"decision": decision,
|
|
"commit_oid": recovered,
|
|
},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
run_root
|
|
).digest,
|
|
now=timestamp,
|
|
)
|
|
record = self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="conflicted",
|
|
state="merged",
|
|
head_commit=recovered,
|
|
last_operation_id=operation_id,
|
|
conflict_interaction_id=interaction_id,
|
|
release_lease=True,
|
|
run_event=self._conflict_resolved_event(
|
|
record,
|
|
interaction_id=interaction_id,
|
|
decision=decision,
|
|
commit_oid=recovered,
|
|
now=timestamp,
|
|
),
|
|
now=timestamp,
|
|
)
|
|
return GitAgentMergeOutcome(
|
|
record,
|
|
"merged",
|
|
recovered,
|
|
(),
|
|
interaction_id,
|
|
)
|
|
|
|
def _materialize(
|
|
self,
|
|
record: GitAgentWorkspaceRecord,
|
|
*,
|
|
run_workspace: GitRunWorkspace,
|
|
lease_token: str,
|
|
operation_request_id: str,
|
|
now: float,
|
|
) -> GitAgentWorkspaceRecord:
|
|
repository = self.journal.get_git_repository(record.repository_id)
|
|
if repository is None:
|
|
raise ContentRepositoryError("Agent repository is unavailable")
|
|
operation_id = (
|
|
"gitop_"
|
|
+ canonical_digest(
|
|
{
|
|
"repository_id": record.repository_id,
|
|
"request_id": f"{operation_request_id}:materialize-agent",
|
|
}
|
|
)[:32]
|
|
)
|
|
with self.content.repository_lock(
|
|
self.content.repository_lock_path(repository.space_id)
|
|
):
|
|
operation = self.journal.begin_git_operation(
|
|
operation_id=operation_id,
|
|
repository_id=record.repository_id,
|
|
request_id=f"{operation_request_id}:materialize-agent",
|
|
operation_type="agent.materialize",
|
|
payload_digest=canonical_digest(
|
|
{
|
|
"workspace_id": record.workspace_id,
|
|
"agent_ref": record.agent_ref,
|
|
"base_commit": record.base_commit,
|
|
}
|
|
),
|
|
expected_repo_state_digest=self.git.repo_state_token(
|
|
Path(repository.root_path)
|
|
).digest,
|
|
)
|
|
if record.state == "admitted":
|
|
record = self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="admitted",
|
|
state="materializing",
|
|
last_operation_id=operation_id,
|
|
now=now,
|
|
)
|
|
if operation.status == "prepared":
|
|
self.journal.mark_git_operation_dispatched(
|
|
operation_id,
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
Path(repository.root_path)
|
|
).digest,
|
|
)
|
|
self.git.ensure_worktree(
|
|
Path(repository.root_path),
|
|
worktree_path=Path(record.worktree_path),
|
|
ref_name=record.agent_ref,
|
|
commit_oid=record.base_commit,
|
|
)
|
|
head = self.git.current_head(Path(record.worktree_path))
|
|
if head is None:
|
|
raise ContentRepositoryError("Agent worktree has no HEAD")
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={"workspace_id": record.workspace_id, "head": head},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
Path(record.worktree_path)
|
|
).digest,
|
|
now=now,
|
|
)
|
|
return self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="materializing",
|
|
state="ready",
|
|
head_commit=head,
|
|
last_operation_id=operation_id,
|
|
now=now,
|
|
)
|
|
|
|
def _refresh_agent_projection(
|
|
self,
|
|
record: GitAgentWorkspaceRecord,
|
|
*,
|
|
run_workspace: GitRunWorkspace,
|
|
lease_token: str,
|
|
run_head: str,
|
|
now: float,
|
|
) -> GitAgentWorkspaceRecord:
|
|
root = Path(record.worktree_path)
|
|
current = self.git.current_head(root)
|
|
if current is None:
|
|
raise ContentRepositoryError("Agent worktree has no HEAD")
|
|
if current != run_head:
|
|
if not self.git.is_worktree_clean(root):
|
|
raise ContentRepositoryError(
|
|
"Agent worktree changed before projection refresh"
|
|
)
|
|
if self.git.is_ancestor(root, run_head, current):
|
|
# The Agent already has checkpointed commits which have not
|
|
# yet been merged into Run integration (for example after a
|
|
# crash between checkpoint and merge). Preserve that durable
|
|
# result for reconciliation instead of resetting it away.
|
|
return self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state=record.state,
|
|
state="ready",
|
|
head_commit=current,
|
|
conflict_interaction_id=None,
|
|
now=now,
|
|
)
|
|
if not self.git.is_ancestor(root, current, run_head):
|
|
raise ContentRepositoryError(
|
|
"Agent branch diverged from Run integration"
|
|
)
|
|
self.git.refresh_owned_worktree(
|
|
root,
|
|
expected_projected_head=current,
|
|
target_head=run_head,
|
|
)
|
|
return self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state=record.state,
|
|
state="ready",
|
|
head_commit=run_head,
|
|
conflict_interaction_id=None,
|
|
now=now,
|
|
)
|
|
|
|
def _recover_dispatched_merge(
|
|
self,
|
|
record: GitAgentWorkspaceRecord,
|
|
*,
|
|
run_workspace: GitRunWorkspace,
|
|
lease_token: str,
|
|
now: float,
|
|
) -> GitAgentWorkspaceRecord:
|
|
operation_id = record.last_operation_id
|
|
recovered = (
|
|
self.git.find_commit_by_operation(
|
|
run_workspace.run_worktree, operation_id
|
|
)
|
|
if operation_id
|
|
else None
|
|
)
|
|
if recovered is not None:
|
|
if operation_id is not None:
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={
|
|
"workspace_id": record.workspace_id,
|
|
"merged_commit": recovered,
|
|
},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
run_workspace.run_worktree
|
|
).digest,
|
|
now=now,
|
|
)
|
|
return self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="merging",
|
|
state="merged",
|
|
head_commit=recovered,
|
|
last_operation_id=operation_id,
|
|
run_event=RunEventDraft(
|
|
event_id=f"git-agent-merge:{operation_id}",
|
|
event_type="git.agent_merged",
|
|
payload={
|
|
"workspace_id": record.workspace_id,
|
|
"agent_id": record.agent_id,
|
|
"commit_oid": recovered,
|
|
"recovered": True,
|
|
},
|
|
created_at=now,
|
|
),
|
|
now=now,
|
|
)
|
|
pending_conflict = next(
|
|
(
|
|
interaction
|
|
for interaction in self.journal.list_human_interactions(
|
|
record.run_id
|
|
)
|
|
if interaction.interaction_type == "merge_conflict"
|
|
and interaction.status in {"requested", "presented"}
|
|
and interaction.request.get("agent_id") == record.agent_id
|
|
),
|
|
None,
|
|
)
|
|
operation = (
|
|
self.journal.get_git_operation(operation_id)
|
|
if operation_id is not None
|
|
else None
|
|
)
|
|
conflict_result = operation.result if operation is not None else None
|
|
if pending_conflict is not None or (
|
|
conflict_result is not None
|
|
and conflict_result.get("status")
|
|
in {"conflicted", "needs_attention"}
|
|
):
|
|
interaction_id = (
|
|
pending_conflict.interaction_id
|
|
if pending_conflict is not None
|
|
else conflict_result.get("interaction_id")
|
|
)
|
|
conflict_paths = tuple(
|
|
str(path)
|
|
for path in (
|
|
pending_conflict.request.get("conflict_paths", [])
|
|
if pending_conflict is not None
|
|
else conflict_result.get("conflict_paths", [])
|
|
)
|
|
)
|
|
state = "conflicted" if interaction_id else "needs_attention"
|
|
if operation_id is not None:
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={
|
|
"workspace_id": record.workspace_id,
|
|
"status": state,
|
|
"conflict_paths": list(conflict_paths),
|
|
"interaction_id": interaction_id,
|
|
},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
run_workspace.run_worktree
|
|
).digest,
|
|
now=now,
|
|
)
|
|
return self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="merging",
|
|
state=state,
|
|
last_operation_id=operation_id,
|
|
conflict_interaction_id=(
|
|
str(interaction_id) if interaction_id else None
|
|
),
|
|
release_lease=True,
|
|
now=now,
|
|
)
|
|
if not self.git.is_worktree_clean(run_workspace.run_worktree):
|
|
raise ContentRepositoryError(
|
|
"Interrupted Run merge left an unclean integration worktree"
|
|
)
|
|
return self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=lease_token,
|
|
expected_state="merging",
|
|
state="ready",
|
|
now=now,
|
|
)
|
|
|
|
def _complete_merge(
|
|
self,
|
|
record: GitAgentWorkspaceRecord,
|
|
*,
|
|
workspace: GitAgentWorkspace,
|
|
operation_id: str,
|
|
merged_commit: str,
|
|
now: float,
|
|
) -> GitAgentMergeOutcome:
|
|
agent_head = self.git.current_head(workspace.agent_worktree)
|
|
if agent_head is None:
|
|
raise ContentRepositoryError("Agent worktree has no HEAD")
|
|
if agent_head != merged_commit:
|
|
self.git.refresh_owned_worktree(
|
|
workspace.agent_worktree,
|
|
expected_projected_head=agent_head,
|
|
target_head=merged_commit,
|
|
)
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={
|
|
"workspace_id": record.workspace_id,
|
|
"merged_commit": merged_commit,
|
|
},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
workspace.run_workspace.run_worktree
|
|
).digest,
|
|
now=now,
|
|
)
|
|
record = self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=workspace.lease_token,
|
|
expected_state="merging",
|
|
state="merged",
|
|
head_commit=merged_commit,
|
|
last_operation_id=operation_id,
|
|
release_lease=True,
|
|
run_event=RunEventDraft(
|
|
event_id=f"git-agent-merge:{operation_id}",
|
|
event_type="git.agent_merged",
|
|
payload={
|
|
"workspace_id": record.workspace_id,
|
|
"agent_id": record.agent_id,
|
|
"commit_oid": merged_commit,
|
|
},
|
|
created_at=now,
|
|
),
|
|
now=now,
|
|
)
|
|
return GitAgentMergeOutcome(record, "merged", merged_commit, (), None)
|
|
|
|
def _record_conflict(
|
|
self,
|
|
record: GitAgentWorkspaceRecord,
|
|
*,
|
|
workspace: GitAgentWorkspace,
|
|
operation_id: str,
|
|
conflict_paths: tuple[str, ...],
|
|
now: float,
|
|
) -> GitAgentMergeOutcome:
|
|
run = self.journal.get_run(record.run_id)
|
|
interaction_id: str | None = None
|
|
if run is not None and run.status not in {
|
|
"completed",
|
|
"failed",
|
|
"cancelled",
|
|
}:
|
|
interaction_id = (
|
|
"merge-conflict:"
|
|
+ canonical_digest(
|
|
{
|
|
"workspace_id": record.workspace_id,
|
|
"operation_id": operation_id,
|
|
"paths": conflict_paths,
|
|
}
|
|
)[:32]
|
|
)
|
|
self.journal.create_human_interaction(
|
|
interaction_id=interaction_id,
|
|
run_id=record.run_id,
|
|
attempt_id=run.active_attempt_id,
|
|
interaction_type="merge_conflict",
|
|
request={
|
|
"title": "Agent changes need conflict resolution",
|
|
"question": (
|
|
f"Agent {record.agent_id} changed files that also "
|
|
"changed in the Run workspace."
|
|
),
|
|
"agent_id": record.agent_id,
|
|
"agent_ref": record.agent_ref,
|
|
"conflict_paths": list(conflict_paths),
|
|
},
|
|
options=[
|
|
{
|
|
"id": "keep_run",
|
|
"label": "Keep Run version",
|
|
},
|
|
{
|
|
"id": "take_agent",
|
|
"label": "Use Agent version",
|
|
},
|
|
{"id": "manual", "label": "Resolve manually"},
|
|
],
|
|
requested_by="workspace_git",
|
|
now=now,
|
|
)
|
|
target_state = "conflicted" if interaction_id else "needs_attention"
|
|
self.journal.complete_git_operation(
|
|
operation_id,
|
|
result={
|
|
"workspace_id": record.workspace_id,
|
|
"status": target_state,
|
|
"conflict_paths": list(conflict_paths),
|
|
"interaction_id": interaction_id,
|
|
},
|
|
observed_repo_state_digest=self.git.repo_state_token(
|
|
workspace.run_workspace.run_worktree
|
|
).digest,
|
|
now=now,
|
|
)
|
|
record = self.journal.transition_git_agent_workspace(
|
|
record.workspace_id,
|
|
lease_token=workspace.lease_token,
|
|
expected_state="merging",
|
|
state=target_state,
|
|
last_operation_id=operation_id,
|
|
conflict_interaction_id=interaction_id,
|
|
release_lease=True,
|
|
now=now,
|
|
)
|
|
return GitAgentMergeOutcome(
|
|
record,
|
|
"conflicted",
|
|
None,
|
|
conflict_paths,
|
|
interaction_id,
|
|
)
|
|
|
|
@staticmethod
|
|
def _conflict_decision(value: dict) -> str:
|
|
decision = value.get("decision")
|
|
if decision is None:
|
|
decision = value.get("option_id", value.get("choice"))
|
|
if decision not in {"keep_run", "take_agent", "manual"}:
|
|
raise ContentRepositoryError(
|
|
"Merge conflict decision must be keep_run, take_agent, or manual"
|
|
)
|
|
return str(decision)
|
|
|
|
@staticmethod
|
|
def _conflict_resolved_event(
|
|
record: GitAgentWorkspaceRecord,
|
|
*,
|
|
interaction_id: str,
|
|
decision: str,
|
|
commit_oid: str | None,
|
|
now: float,
|
|
) -> RunEventDraft:
|
|
return RunEventDraft(
|
|
event_id=f"git-agent-conflict:{interaction_id}:{decision}",
|
|
event_type="git.agent_conflict_resolved",
|
|
payload={
|
|
"workspace_id": record.workspace_id,
|
|
"agent_id": record.agent_id,
|
|
"interaction_id": interaction_id,
|
|
"decision": decision,
|
|
"commit_oid": commit_oid,
|
|
},
|
|
created_at=now,
|
|
)
|
|
|
|
@staticmethod
|
|
def _segment(*values: str) -> str:
|
|
return canonical_digest(list(values))[:24]
|
|
|
|
def _agent_ref(
|
|
self, repository_id: str, run_id: str, agent_id: str
|
|
) -> str:
|
|
return (
|
|
# Run integration already owns refs/heads/eigent/run/<run> as a
|
|
# leaf. Git cannot create child refs below an existing leaf, so
|
|
# Agent branches live in a sibling namespace.
|
|
"refs/heads/eigent/agent/"
|
|
+ self._segment(repository_id, run_id)
|
|
+ "/"
|
|
+ self._segment(agent_id)
|
|
)
|
|
|
|
def _agent_worktree(
|
|
self, space_id: str, run_id: str, agent_id: str
|
|
) -> Path:
|
|
return (
|
|
self.state_root
|
|
/ "worktrees"
|
|
/ self._segment("space", space_id)
|
|
/ "runs"
|
|
/ self._segment("run", run_id)
|
|
/ "agents"
|
|
/ self._segment("agent", agent_id)
|
|
)
|
|
|
|
|
|
def get_default_workforce_git_service() -> WorkforceGitService:
|
|
return WorkforceGitService(
|
|
get_default_run_journal(),
|
|
state_root=configured_run_journal_path().parent / "workspace-git",
|
|
)
|