並列エージェントが同じファイルに書き込むときの取り合いを防ぐ

from dataclasses import dataclass, field
from fnmatch import fnmatch
 
@dataclass
class AgentScope:
    agent_id: str
    owned_globs: list[str]          # 書き込み許可パターン
    shared_readonly: list[str] = field(default_factory=list)
 
    def can_write(self, path: str) -> bool:
        # 所有グロブに一致すれば書き込み可。それ以外は拒否
        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} は {path} の所有権を持ちません")

import json, queue, threading
 
write_queue: "queue.Queue[dict]" = queue.Queue()
 
def request_shared_edit(file: str, mutate) -> None:
    # mutate: 現在の内容(dict)を受け取り新しい内容を返す純粋関数
    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:        # 直前に読み直す = 最新状態を反映
            current = json.load(f)
        updated = job["mutate"](current)     # 各要求を順に適用
        with open(job["file"], "w") as f:
            json.dump(updated, f, indent=2, ensure_ascii=False)
        write_queue.task_done()

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()
            # 空き、または期限切れのリースなら奪取できる
            if holder is None or holder[1] < now:
                self._leases[path] = (agent_id, now + self._ttl)
                return True
            return holder[0] == agent_id   # 自分が保持中なら再入可
 
    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]

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:
        # 自分の書き込み後に第三者が触れた = ロストアップデートの疑い
        raise RuntimeError(f"{path} が想定外に変化 expected={expected} actual={actual}")