fix(agent): per-task git worktree so concurrent tasks each get an isolated repo copy
取代上一版 Option B(执行器共用仓库根 + 串行 1 任务)。每个任务从本 agent 自己的 clone 切出独立 git worktree(完整仓库内容 + 独立分支/索引),执行器在其中读写真实源码, 任务级 GitOperations 在该 worktree 提交/推送结果分支,完成后回收 worktree。 - 单 agent 可并发多任务(MAX_CONCURRENT_TASKS 恢复 4),互不共用 checkout/index - worktree 置于仓外 /tmp/agent-worktrees(AGENT_WORKTREE_BASE 可配),主 checkout 不受污染 - _git_admin_lock 仅串行 worktree add/remove 等共享 .git plumbing,任务执行仍并行 - 跨 agent 隔离不变:每个 agent 仍各自 clone 一份(各自 pod) - 根治 "Empty workspace: no files detected"(worktree 自带仓库文件,已端到端验证) GitOperations 新增 add_task_worktree / remove_task_worktree。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
gongzhiyong
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
6152235c96
commit
62d6fdf78c
@@ -164,6 +164,46 @@ class GitOperations:
|
||||
logger.error(f"Error creating result branch: {e}")
|
||||
return False
|
||||
|
||||
async def add_task_worktree(self, path: str, task_id: str) -> Optional[str]:
|
||||
"""Create an isolated git worktree at `path` on a fresh per-task result branch.
|
||||
|
||||
Cut from this agent's clone (off origin/<base> or HEAD), the worktree has the full repo
|
||||
contents plus its own index/HEAD, so one agent can run several tasks concurrently without
|
||||
them clobbering a shared checkout. Returns the branch name, or None on failure.
|
||||
"""
|
||||
try:
|
||||
root = await self.repo_root() or self.workspace_dir
|
||||
safe_task_id = task_id.replace("/", "-")[:12]
|
||||
branch = f"agent/{self.agent_id}/{safe_task_id}-{int(time.time())}"
|
||||
|
||||
await self._run_git_command(["git", "fetch", "origin"], cwd=root)
|
||||
base_ref = f"origin/{self.base_branch}"
|
||||
if (await self._run_git_command(["git", "rev-parse", "--verify", base_ref], cwd=root)).returncode != 0:
|
||||
base_ref = "HEAD"
|
||||
logger.warning(f"Base branch origin/{self.base_branch} not found; worktree from HEAD")
|
||||
|
||||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
||||
result = await self._run_git_command(
|
||||
["git", "worktree", "add", "-b", branch, path, base_ref], cwd=root,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
logger.error(f"git worktree add failed: {result.stderr}")
|
||||
return None
|
||||
logger.info(f"Created task worktree {path} on branch {branch}")
|
||||
return branch
|
||||
except Exception as e:
|
||||
logger.error(f"Error creating task worktree: {e}")
|
||||
return None
|
||||
|
||||
async def remove_task_worktree(self, path: str) -> None:
|
||||
"""Tear down a per-task worktree (best-effort). The branch ref is kept (already pushed)."""
|
||||
try:
|
||||
root = await self.repo_root() or self.workspace_dir
|
||||
await self._run_git_command(["git", "worktree", "remove", "--force", path], cwd=root)
|
||||
logger.info(f"Removed task worktree {path}")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to remove task worktree {path}: {e}")
|
||||
|
||||
async def commit_changes(self, message: str) -> Optional[str]:
|
||||
try:
|
||||
cwd = await self.repo_root() or self.workspace_dir
|
||||
|
||||
+38
-24
@@ -72,11 +72,10 @@ class AgentRuntimeDisconnected(RuntimeError):
|
||||
class Agent:
|
||||
"""Agent that connects to orchestrator and executes tasks."""
|
||||
|
||||
# One task at a time per agent: tasks execute against the shared cloned repo root (see
|
||||
# _execute_assignment, Option B 2026-06-16) and the per-agent git checkout (workspace_git,
|
||||
# branch/commit/push) is single-repo and not concurrency-safe. Swarm parallelism comes from
|
||||
# MULTIPLE agents, not multiple tasks per agent. Override via env only on isolated workspaces.
|
||||
MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "1"))
|
||||
# An agent may run several tasks at once; each executes in its OWN git worktree cut from this
|
||||
# agent's clone (see _execute_assignment / GitOperations.add_task_worktree), so concurrent tasks
|
||||
# never share a checkout/index. Swarm parallelism = many agents × this per-agent concurrency.
|
||||
MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "4"))
|
||||
TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "60"))
|
||||
PEER_REPLY_TIMEOUT_SECONDS = int(os.getenv("PEER_REPLY_TIMEOUT_SECONDS", "20"))
|
||||
HEARTBEAT_INTERVAL_SECONDS = 15
|
||||
@@ -115,14 +114,20 @@ class Agent:
|
||||
self._peer_executor: Optional[TaskExecutor] = None
|
||||
|
||||
self.workspace_git = GitOperations(str(self.workspace_dir), self.agent_id)
|
||||
# Serializes only the brief shared-repo git plumbing (worktree add/remove + fetch) so
|
||||
# concurrent tasks don't race on .git locks. Task EXECUTION stays parallel.
|
||||
self._git_admin_lock = asyncio.Lock()
|
||||
|
||||
def available_slots(self) -> int:
|
||||
# Return how many additional tasks this agent can currently accept.
|
||||
return max(0, self.MAX_CONCURRENT_TASKS - len(self.active_tasks))
|
||||
|
||||
def task_workspace(self, task_id: str) -> Path:
|
||||
# Compute the dedicated per-task working directory inside the agent workspace.
|
||||
return self.workspace_dir / ".agent_tasks" / task_id
|
||||
# Per-task git worktree path. Kept OUTSIDE the repo root (default /tmp/agent-worktrees) so
|
||||
# the main checkout never sees it as untracked, and namespaced by agent_id to avoid
|
||||
# collisions. `git worktree add` requires the leaf to not pre-exist, so we don't mkdir it.
|
||||
base = Path(os.getenv("AGENT_WORKTREE_BASE", "/tmp/agent-worktrees"))
|
||||
return base / self.agent_id / task_id.replace("/", "-")
|
||||
|
||||
async def safe_send(self, payload: dict):
|
||||
# Serialize and send a websocket message while holding a lock to prevent concurrent writes.
|
||||
@@ -386,22 +391,28 @@ class Agent:
|
||||
await self.send_status_update("busy", task_id, "Starting task execution")
|
||||
|
||||
start_time = time.time()
|
||||
task_workspace = self.task_workspace(task_id)
|
||||
|
||||
git_enabled = False
|
||||
task_git = None
|
||||
try:
|
||||
# Execute against the cloned repository root (Option B, 2026-06-16). The previous
|
||||
# per-task subdir (/workspace/.agent_tasks/<task_id>) was created empty and never
|
||||
# seeded with the repo, so the executor saw an empty workspace and every SWE task
|
||||
# failed with "Empty workspace: no files detected". Using the repo root means the
|
||||
# model reads the real source and git (workspace_git, rooted at the same dir)
|
||||
# commits the actual edits. One-task-per-agent (MAX_CONCURRENT_TASKS) keeps this safe.
|
||||
executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=str(self.workspace_dir))
|
||||
git_enabled = await self.workspace_git.is_git_workspace()
|
||||
if git_enabled:
|
||||
branch_created = await self.workspace_git.create_result_branch(task_id)
|
||||
if not branch_created:
|
||||
logger.warning("Failed to create task result branch; continuing without git push")
|
||||
git_enabled = False
|
||||
# Each task runs in its OWN git worktree cut from this agent's clone, so an agent can
|
||||
# run several tasks concurrently without sharing a checkout/index. The worktree holds
|
||||
# the full repo contents on a fresh result branch; the executor reads/writes the real
|
||||
# source there and the per-task GitOperations (task_git) commits/pushes that branch.
|
||||
# This fixes the prior empty per-task subdir → "Empty workspace: no files detected".
|
||||
# Falls back to the repo root (no isolated branch) only if worktree creation fails.
|
||||
if await self.workspace_git.is_git_workspace():
|
||||
async with self._git_admin_lock: # serialize shared-repo git plumbing only
|
||||
result_branch = await self.workspace_git.add_task_worktree(str(task_workspace), task_id)
|
||||
if result_branch:
|
||||
git_enabled = True
|
||||
task_git = GitOperations(str(task_workspace), self.agent_id)
|
||||
task_git.result_branch = result_branch
|
||||
else:
|
||||
logger.warning("Failed to create task worktree; executing on repo root without git push")
|
||||
exec_dir = str(task_workspace) if git_enabled else str(self.workspace_dir)
|
||||
executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=exec_dir)
|
||||
|
||||
result = await asyncio.wait_for(
|
||||
executor.execute_task(
|
||||
@@ -409,7 +420,7 @@ class Agent:
|
||||
description=description,
|
||||
context={
|
||||
**context,
|
||||
"workspace_dir": str(self.workspace_dir),
|
||||
"workspace_dir": exec_dir,
|
||||
"repo_workspace_dir": str(self.workspace_dir),
|
||||
"git_repo_url": self.git_repo_url,
|
||||
"agent_id": self.agent_id,
|
||||
@@ -425,12 +436,12 @@ class Agent:
|
||||
if result.get("success"):
|
||||
self.last_summary = self._summarize_result(result) or self.last_summary
|
||||
if result.get("success") and not awaiting_handoff:
|
||||
if git_enabled:
|
||||
commit_sha = await self.workspace_git.commit_changes(
|
||||
if git_enabled and task_git:
|
||||
commit_sha = await task_git.commit_changes(
|
||||
message=f"Task {task_id}: {description[:50]}"
|
||||
)
|
||||
if commit_sha:
|
||||
branch_name = await self.workspace_git.push_results()
|
||||
branch_name = await task_git.push_results()
|
||||
result["git_branch"] = branch_name
|
||||
result["commit_sha"] = commit_sha
|
||||
else:
|
||||
@@ -473,6 +484,9 @@ class Agent:
|
||||
await self.send_task_result(task_id, False, {"error": str(e), "success": False})
|
||||
await self.send_status_update("idle", None, f"Task failed: {e}")
|
||||
finally:
|
||||
if git_enabled:
|
||||
async with self._git_admin_lock:
|
||||
await self.workspace_git.remove_task_worktree(str(task_workspace))
|
||||
self.active_tasks.pop(task_id, None)
|
||||
ACTIVE_TASKS.set(len(self.active_tasks))
|
||||
self.current_task_id = None
|
||||
|
||||
Reference in New Issue
Block a user