What is cosm / topocosm?

Go / Python

cosm and topocosm: source control for AI coding agents

When several coding agents edit the same repo at once, git falls over fast. cosm lets each agent lock just the function or class it is changing, not the whole file.

Diagram of cosm and topocosm AST-graph source control architecture showing symbol-level locks, 2-hop blast-radius analysis, and file materialization for multi-agent engineering by Jason Davenport. — cosm and topocosm: each agent locks one function or class instead of a whole file.

What it does

cosm / topocosm in plain terms

cosm stores your code as a graph of functions, classes, and the calls between them instead of as files. Before editing, an agent asks what else the change could break, takes a short lock on just that piece, and cosm writes normal files back out so your usual build and test tools keep working. topocosm is the shared server that coordinates many agents working in one codebase.

How it works

Three things to know

The code graph is the source of truth: every function, class, and the calls between them live in a database.
Before editing, an agent checks what depends on a symbol (two steps out) and takes a short lock on it. Other agents keep working on everything else.
Normal files are generated from the graph, so `go test`, `uv run pytest`, and your build still work.

Sample result: 78.4% fewer tokens in a sample run.

Learn more

Essays and source

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 a code sample (cosm_symbol_lease.py) Go / Python

File: cosm_symbol_lease.py

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(),
        }

Run it: $ cosm universe inspect --symbol 'Router.Dispatch' --blast-radius --depth 2

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%