Stopping Parallel Agents from Clobbering the Same File

Strategy 1: Ownership partitioning (try this first)

The sturdiest move is to never let two agents touch the same file at all. When you split the task, partition the files into per-agent ownership.

Concretely, before the orchestrator issues subtasks it hands each agent a set of writable path globs and forbids writes outside them.

from dataclasses import dataclass, field
from fnmatch import fnmatch
 
@dataclass
class AgentScope:
    agent_id: str
    owned_globs: list[str]          # patterns this agent may write
    shared_readonly: list[str] = field(default_factory=list)
 
    def can_write(self, path: str) -> bool:
        # writable only if the path matches an owned glob
        return any(fnmatch(path, g) for g in self.owned_globs)
 
def assign_scopes() -> list[AgentScope]:
    return [
        AgentScope("ui",   ["src/components/**", "src/styles/**"]),
        AgentScope("api",  ["src/app/api/**", "src/lib/server/**"]),
        AgentScope("test", ["tests/**", "e2e/**"]),
    ]
 
def guard_write(scopes: dict[str, AgentScope], agent_id: str, path: str) -> None:
    if not scopes[agent_id].can_write(path):
        raise PermissionError(f"{agent_id} does not own {path}")

Drop guard_write into the agent's file-write hook and any out-of-scope overwrite stops before it runs. If your Antigravity agent config lets you restrict writable directories, use that as the first line of defense and keep this guard as the second.

Ownership partitioning runs out of road on shared files everyone wants to touch, like package.json or type definitions. The next two strategies handle those.

Strategy 2: A serialization point for shared files

Changes to shared files are not written directly. Each agent enqueues a "change request," and a single dedicated agent (or the orchestrator itself) applies them one at a time.

import json, queue, threading
 
write_queue: "queue.Queue[dict]" = queue.Queue()
 
def request_shared_edit(file: str, mutate) -> None:
    # mutate: a pure function that takes current content (dict) and returns the new content
    write_queue.put({"file": file, "mutate": mutate})
 
def serializer_loop(stop: threading.Event) -> None:
    while not stop.is_set():
        try:
            job = write_queue.get(timeout=0.5)
        except queue.Empty:
            continue
        with open(job["file"]) as f:        # re-read right before applying = latest state
            current = json.load(f)
        updated = job["mutate"](current)     # apply each request in order
        with open(job["file"], "w") as f:
            json.dump(updated, f, indent=2, ensure_ascii=False)
        write_queue.task_done()

The key is making mutate a function that takes the current content and returns new content. Each agent sends only its intent ("add one dependency"), and the serializer re-reads the latest state right before applying it, so lost updates cannot happen by construction. Multiple agents appending to package.json settle most cleanly with this approach.

Strategy 3: A lease-based file ownership coordinator

For the middle ground — where you cannot statically partition ownership but a full serialization queue is overkill — a short-lived lease (a time-boxed lock) fits. Acquire before writing, release when done; if you cannot acquire, wait or move to another task.

import time, threading
 
class FileLeaseCoordinator:
    def __init__(self, lease_ttl: float = 30.0):
        self._lock = threading.Lock()
        self._leases: dict[str, tuple[str, float]] = {}  # path -> (agent, expiry)
        self._ttl = lease_ttl
 
    def acquire(self, path: str, agent_id: str) -> bool:
        with self._lock:
            holder = self._leases.get(path)
            now = time.monotonic()
            # free, or an expired lease, means we can take it
            if holder is None or holder[1] < now:
                self._leases[path] = (agent_id, now + self._ttl)
                return True
            return holder[0] == agent_id   # re-entrant if we already hold it
 
    def release(self, path: str, agent_id: str) -> None:
        with self._lock:
            if self._leases.get(path, (None, 0))[0] == agent_id:
                del self._leases[path]

The TTL exists so that if an agent crashes without releasing, the file is not locked forever. I got burned here once. I started with an unbounded lock, and the moment one agent stalled on a timeout, everyone else froze waiting on that file. In production, always set an expiry and let expired leases be taken over.

Choosing among the three

1. Try ownership partitioning first (never let two agents touch one file)
2. Route truly shared files through a serialization point
3. Handle only dynamic, short-lived contention with leases

A rough decision table:

In my case I settled on three layers: partition roughly 80% of files statically, funnel the remaining shared files into a serialization point, and limit leases to genuinely dynamic temp files. Make leases the lead actor and waiting time grows until the parallelism stops paying off. Keep them as a helper — that is my personal recommendation.

Verification: post a sentry that detects clobbering

Even with the design in place, whether there is a gap is a separate question. In production I added a sentry that records a file's hash after writing and checks whether the value I wrote matches the value read next.

import hashlib
 
def fingerprint(path: str) -> str:
    with open(path, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()[:12]
 
def verify_no_clobber(path: str, expected: str) -> None:
    actual = fingerprint(path)
    if actual != expected:
        # someone touched the file after my write = suspected lost update
        raise RuntimeError(f"{path} changed unexpectedly expected={expected} actual={actual}")

Wrap this right after an agent's write and right before commit, and changes that would have vanished silently surface early as a "broken build" instead. The undetectable lost update is the biggest trap, so I find this sentry worth as much as the arbitration design itself.

Parallel agents are appealing, but if you just add workers without designing arbitration for shared resources, you gain speed and quietly lose work in equal measure. Ownership partitioning as the first defense, serialization and leases as helpers, and a sentry to verify. That ordering is the setup I actually rely on in my own indie development today. I hope it gives you a foothold if you are facing the same wall.