AST-Graph Native Source Control & Symbol Blast-Radius Locking
What happens when 10 coding agents edit the same repo at once? Test how symbol-level AST locks prevent merge collisions and cut context tokens by up to 78%.
Read the essay: WIBTM: What happens when you let an agent design a new git system? →
Related: How to Track AI Agent Lineage and Manage State in Code Repositories →
Related: 100% of your developers should be building things to improve the workflow. Are they? →
GitHub: cosm / topocosm ↗
View source (cosm_symbol_lease.py) and notes
Flips traditional file based Git on its head for multi-agent engineering: the Abstract Syntax Tree (AST) graph is the primary source of truth, symbol-level leases prevent concurrent agent merge collisions, and files are materialized downstream for build tool compatibility.
Design notes
- Primary Ground Truth: AST entities and directed call/type edges stored in a transactional graph universe.
- Blast-Radius Analysis: Agents query 2-hop symbol dependencies before acquiring a short lived surgical lease.
- Secondary File Materialization: Language renderers project the graph back to disk for `go test` or `uv run pytest`.
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
@dataclass(frozen=True)
class ASTSymbolNode:
symbol_id: str
qualified_name: str
kind: str # "Function" | "Method" | "Struct" | "Interface"
signature_hash: str
dependencies: tuple[str, ...] = field(default_factory=tuple)
class CosmUniverseGraph:
"""Agent-first AST graph universe with symbol-scoped locking and blast-radius traversal."""
def __init__(self, universe_name: str) -> None:
self.universe_name = universe_name
self.nodes: dict[str, ASTSymbolNode] = {}
self.active_leases: dict[str, tuple[str, datetime]] = {}
def register_symbol(self, node: ASTSymbolNode) -> None:
self.nodes[node.symbol_id] = node
def compute_blast_radius(self, target_symbol: str, max_hops: int = 2) -> list[str]:
"""Return all upstream symbols impacted if `target_symbol` mutates."""
reverse_edges: dict[str, set[str]] = {sid: set() for sid in self.nodes}
for sid, node in self.nodes.items():
for dep in node.dependencies:
reverse_edges.setdefault(dep, set()).add(sid)
visited: list[str] = []
frontier = {target_symbol}
for _ in range(max_hops):
next_hop: set[str] = set()
for current in frontier:
for caller in reverse_edges.get(current, set()):
if caller not in visited and caller != target_symbol:
visited.append(caller)
next_hop.add(caller)
frontier = next_hop
return visited
def acquire_surgical_lease(self, symbol_id: str, agent_id: str, ttl_seconds: int = 30) -> dict[str, object]:
now = datetime.now(timezone.utc)
lease = self.active_leases.get(symbol_id)
if lease and lease[1] > now and lease[0] != agent_id:
raise RuntimeError(f"Symbol {symbol_id} locked by {lease[0]} until {lease[1].isoformat()}")
expires = now + timedelta(seconds=ttl_seconds)
self.active_leases[symbol_id] = (agent_id, expires)
return {
"universe": self.universe_name,
"symbol_id": symbol_id,
"agent_id": agent_id,
"blast_radius": self.compute_blast_radius(symbol_id),
"expires_at": expires.isoformat(),
}
UNIVERSE: davenportjw/cosm-core (rev: 9f41c8a2) LOCK STATUS: AVAILABLE (0 active agent leases) SYMBOL: Router.Dispatch (kind: Method, lang: Go) DIRECT CALLERS (depth=1): [Gateway.HandleRequest, ShardWorker.SyncTick] TRANSITIVE IMPACT (depth=2): 4 symbols across 2 modules · Token savings vs full-file read: 78.4%
