Preventing Multi-Agent Code Conflicts and Token Explosion with Git Worktree and AST Analysis
makedreamJuly 26, 2026
0
Computing/Software
Comments (0)
Log in to leave a comment
No posts yet
makedreammakedreamLog in to leave a comment
No posts yet
When you scale LLM agents that worked well in demos up to a swarm level, you inevitably hit two major walls. Either agents overwrite each other's files, tangling up the codebase, or you shovel irrelevant files into large models, burning millions of won in API costs.
Running multiple language model processes indiscriminately in the same workspace quickly escalates the situation. Agent B reads an incomplete file while Agent A is modifying it, generating bizarre code, and eventually even the commit history gets wiped out. On the other hand, copying the entire repository every time wastes disk space and takes minutes just for initialization.
Here is how to solve this problem through engineering by combining in-memory file isolation, syntax-analysis-based routing, and a static validation pipeline.
The bottleneck that occurs when running agents concurrently on a large codebase is file system contention. Instead of doing a Full Git Clone of an entire monolithic repository, using Git Worktree lets you share metadata and the object database while isolating lightweight directories at the level of a few MBs in just 1 second.
However, when dozens of agents make commits simultaneously, lock contention occurs on the upper index file (.git/index.lock). To control this, a sandbox layer based on file locking is required.
`python
import os
import sys
import time
import subprocess
import shutil
from pathlib import Path
from typing import Optional, List
from filelock import FileLock, Timeout
class WorktreeSandboxManager:
def init(self, repo_path: str, base_branch: str = "main"):
self.repo_path = Path(repo_path).resolve()
self.base_branch = base_branch
self.worktrees_dir = self.repo_path / ".agent_worktrees"
self.locks_dir = self.repo_path / ".agent_locks"
self.worktrees_dir.mkdir(exist_ok=True)
self.locks_dir.mkdir(exist_ok=True)
def create_sandbox(self, agent_id: str, task_name: str) -> Path:
branch_name = f"agent/{agent_id}-{task_name}"
worktree_path = self.worktrees_dir / f"wt_{agent_id}"
if worktree_path.exists():
self.cleanup_sandbox(agent_id, force=True)
cmd = [
"git", "-C", str(self.repo_path),
"worktree", "add", "-b", branch_name,
str(worktree_path), self.base_branch
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Worktree creation failed: {result.stderr}")
return worktree_path
def safe_git_commit(self, worktree_path: Path, commit_message: str, max_retries: int = 5) -> bool:
lock_file_path = self.locks_dir / "git_index.lock"
file_lock = FileLock(str(lock_file_path), timeout=10)
for attempt in range(max_retries):
try:
with file_lock:
add_res = subprocess.run(
["git", "-C", str(worktree_path), "add", "."],
capture_output=True, text=True
)
if add_res.returncode != 0:
raise RuntimeError(f"Git add failed: {add_res.stderr}")
commit_res = subprocess.run(
["git", "-C", str(worktree_path), "commit", "-m", commit_message],
capture_output=True, text=True
)
if commit_res.returncode == 0:
return True
if "index.lock" in commit_res.stderr or "Unable to create" in commit_res.stderr:
backoff = (2 ** attempt) * 0.2
time.sleep(backoff)
continue
else:
print(f"Commit failed (non-contention error): {commit_res.stderr}")
return False
except (Timeout, RuntimeError) as e:
backoff = (2 ** attempt) * 0.2
time.sleep(backoff)
return False
def cleanup_sandbox(self, agent_id: str, force: bool = False):
worktree_path = self.worktrees_dir / f"wt_{agent_id}"
if not worktree_path.exists():
return
status_res = subprocess.run(
["git", "-C", str(worktree_path), "status", "--porcelain"],
capture_output=True, text=True
)
if status_res.stdout.strip() and not force:
raise RuntimeError("Cannot remove Worktree because uncommitted changes exist.")
subprocess.run(
["git", "-C", str(self.repo_path), "worktree", "remove", "--force", str(worktree_path)],
capture_output=True, text=True
)
if worktree_path.exists():
shutil.rmtree(worktree_path, ignore_errors=True)
`
The application sequence is straightforward:
filelock package and add the WorktreeSandboxManager class to your project.create_sandbox() to spin up an isolated directory.safe_git_commit() to avoid lock conflicts via exponential backoff.Changing the configuration like this eliminates overwrite conflicts. Time wasted on debugging also drops by more than 5 hours per week.
When bringing modified branches back into the main codebase, you should use Abstract Syntax Tree (AST) analysis instead of line-by-line text merging to remain safe. Simple text merging throws conflicts even if only the top import statements change position. Parsing source code into a syntax node tree using Python's built-in ast module or Tree-Sitter, and then merging at the function or class level, reduces merge failure rates to nearly 0%.
Attaching Claude 3.5 Sonnet to every task indiscriminately makes costs unbearable. You shouldn't gauge complexity purely by lines of code (LOC). A 100-line piece of code packed with convoluted ternary operators and nested conditionals is far more challenging than a 500-line data class filled with comments.
By using the ast module, you can calculate node count, cyclomatic complexity, and tree depth, converting them into an objectified score.
`python
import ast
class CodeComplexityAnalyzer(ast.NodeVisitor):
def init(self):
self.node_count = 0
self.max_depth = 0
self.current_depth = 0
self.cyclomatic_complexity = 1
def generic_visit(self, node):
self.node_count += 1
self.current_depth += 1
if self.current_depth > self.max_depth:
self.max_depth = self.current_depth
super().generic_visit(node)
self.current_depth -= 1
def visit_If(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_For(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_While(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_ExceptHandler(self, node):
self.cyclomatic_complexity += 1
self.generic_visit(node)
def visit_BoolOp(self, node):
self.cyclomatic_complexity += len(node.values) - 1
self.generic_visit(node)
def calculate_ast_metrics(source_code: str) -> dict:
try:
tree = ast.parse(source_code)
analyzer = CodeComplexityAnalyzer()
analyzer.visit(tree)
score = (analyzer.node_count * 0.2) + (analyzer.max_depth * 1.5) + (analyzer.cyclomatic_complexity * 3.0)
return {
"node_count": analyzer.node_count,
"max_depth": analyzer.max_depth,
"cyclomatic_complexity": analyzer.cyclomatic_complexity,
"complexity_score": round(score, 2),
"is_valid": True
}
except SyntaxError as e:
return {"is_valid": False, "error": str(e), "complexity_score": 9999}
`
Place this analyzer at the entry point of your backend pipeline and set the routing threshold score to 50.
Tasks scoring below 50—such as writing unit tests, implementing utilities, or defining DTOs—are offloaded to Claude 3.5 Haiku, which costs around $0.80 per million input tokens. Only large-scale refactoring or structural design scoring 50 or higher are routed to Claude 3.5 Sonnet at $3.00 per million tokens. Having Haiku handle over 60% of total traffic alone slashes API costs by up to 60%.
Token bleeding caused by long conversation contexts can be cut off using a session reset middleware. When tracking accumulated tokens reaches a threshold, forcibly reset the conversation. At this point, create a summary using AST to extract key function symbols and remaining TODOs, and feed it as the initial prompt of the new session, allowing work to continue seamlessly without context loss.
Merging draft code generated by agents directly into the repository will break the build. On the other hand, calling the LLM again just to catch simple typos or syntax errors is both time-consuming and wasteful.
Construct a step-by-step validation pipeline pairing linters, type checkers, and an LLM reviewer.
`python
import ast
import subprocess
from pathlib import Path
from typing import Optional
from pydantic import BaseModel, Field
class ValidationResult(BaseModel):
is_success: bool = Field(description="Validation pass status")
failed_stage: Optional[str] = Field(default=None, description="Failed validation stage")
error_message: Optional[str] = Field(default=None, description="Error message")
suggested_context: Optional[str] = Field(default=None, description="Context to inject for fixes")
class MultiLensReviewerChain:
def init(self, worktree_path: Path):
self.worktree_path = worktree_path
def run_stage1_ast_lint(self, file_path: Path) -> ValidationResult:
try:
with open(file_path, "r", encoding="utf-8") as f:
code_content = f.read()
ast.parse(code_content)
except SyntaxError as e:
return ValidationResult(
is_success=False,
failed_stage="Stage 1 (AST Syntax)",
error_message=f"SyntaxError occurred at line {e.lineno}: {e.msg}",
suggested_context=e.text
)
res = subprocess.run(["ruff", "check", str(file_path)], capture_output=True, text=True)
if res.returncode != 0:
return ValidationResult(
is_success=False,
failed_stage="Stage 1 (Ruff Linter)",
error_message=res.stdout or res.stderr
)
return ValidationResult(is_success=True)
def run_stage2_type_check(self, file_path: Path) -> ValidationResult:
res = subprocess.run(
["mypy", "--config-file", "mypy.ini", str(file_path)],
capture_output=True, text=True, cwd=str(self.worktree_path)
)
if res.returncode != 0:
return ValidationResult(
is_success=False,
failed_stage="Stage 2 (Mypy TypeChecker)",
error_message=res.stdout
)
return ValidationResult(is_success=True)
def execute_pipeline(self, target_file_rel_path: str) -> ValidationResult:
full_path = self.worktree_path / target_file_rel_path
s1_res = self.run_stage1_ast_lint(full_path)
if not s1_res.is_success:
return s1_res
s2_res = self.run_stage2_type_check(full_path)
if not s2_res.is_success:
return s2_res
return ValidationResult(is_success=True)
`
Stage 1 catches syntax errors with AST parsing and Ruff, while Stage 2 aligns types with Mypy. Only code that passes all these static validation tools is sent to the Stage 3 Claude 3.5 Sonnet deep reviewer. Re-invoking the LLM due to missing parentheses or simple type errors disappears, speeding up pipeline completion time by 40%.
To prevent getting trapped in a loop upon validation failure, a circuit breaker is essential. Limit retries for the same error to a maximum of 3 times, and if the error message hash value is identical to the previous one, judge that the agent has fallen into a hallucination loop and immediately abort execution.
To centrally manage which files and branches multiple agents are touching, you need at least a basic SQLite schema.
`sql
CREATE TABLE agent_sessions (
agent_id TEXT PRIMARY KEY,
worktree_path TEXT NOT NULL,
current_status TEXT CHECK(current_status IN ('IDLE', 'RUNNING', 'LINTING', 'FAILED', 'COMPLETED')),
assigned_task TEXT,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE file_locks (
file_path TEXT PRIMARY KEY,
locked_by_agent TEXT NOT NULL,
ast_symbol_node TEXT,
lock_acquired_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY(locked_by_agent) REFERENCES agent_sessions(agent_id)
);
CREATE TABLE context_events (
event_id INTEGER PRIMARY KEY AUTOINCREMENT,
source_agent TEXT NOT NULL,
event_type TEXT CHECK(event_type IN ('FILE_MUTATED', 'INTERFACE_CHANGED', 'ROLLBACK_TRIGGERED')),
affected_path TEXT NOT NULL,
payload_json TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
`
Every time an agent succeeds in committing to a shared module, it emits a FILE_MUTATED event. Other agents receive this notification and instantly update their referenced AST symbol definitions to the latest version.
If a specific agent enters an unrecoverable state after a validation failure, execute an atomic rollback using the snapshot commit SHA taken when starting the task.
bash git -C .agent_worktrees/wt_agent_01 reset --hard <SNAPSHOT_COMMIT_SHA> git -C .agent_worktrees/wt_agent_01 clean -fd
By tying together isolated directories, syntax-based model routing, static validation pipelines, and state databases like this, you can reliably run a production-grade agent swarm without worrying about file conflicts or runaway costs.