#!/usr/bin/env python3 """echo_concurrency.py — H3: make the multi-writer story actually safe. [v1.0 SCAFFOLD — not yet wired] The plugin advertises a shared vault (Claude Code + CoWork), but two real races exist: 1. The entity index is a full-file load -> mutate -> PUT (echo_index.load/save) with NO lock. Two concurrent `capture` calls both read the old index; last PUT wins; the other entity is silently dropped from the index (the note file survives, but resolve/recall miss it until the next sweep). 2. The advisory lock is MANUAL (echo.py cmd_lock) — capture / scope set never take it; it relies on the model remembering to. This module provides (a) a context manager that auto-acquires/releases the existing advisory lock around a write burst, with crash-safe release, and (b) an atomic index-update that re-reads UNDER the lock before saving, so concurrent captures merge instead of clobber. ACCEPTANCE (v1.0 roadmap H3, shipped; roadmap retired — see git history): - two concurrent captures both land in the index (no lost entries) - lock auto-released on normal AND error exit """ from __future__ import annotations import contextlib import os import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import echo # noqa: E402 import echo_index as idx_mod # noqa: E402 def auto_owner() -> str: """A stable-per-process owner id: client tag + pid. Avoids Math.random/time-based ids; pid is unique enough for cooperative locking and is reproducible in tests.""" tag = os.environ.get("ECHO_CLIENT", "cc") return f"{tag}-{os.getpid()}" @contextlib.contextmanager def vault_lock(owner: str | None = None, required: bool = False): """Hold the advisory lock for a write burst; release on exit (incl. exceptions). Yields `held: bool`. The lock is cooperative (read-back-confirmed, TTL-reclaimable), so if another session holds it we do NOT block: with `required=False` we yield `held=False` and let the caller proceed (advisory) or warn; with `required=True` we raise. Release is best-effort and never masks the body's own exception. """ owner = owner or auto_owner() rc = echo.cmd_lock(owner, quiet=True) held = rc == 0 if not held and required: raise echo.EchoError(f"vault_lock: could not acquire lock for '{owner}' (held elsewhere)", 75) try: yield held finally: if held: try: echo.cmd_unlock(owner, quiet=True) except Exception: # noqa: BLE001 — release is best-effort; TTL reclaims a leak pass def atomic_index_update(mutate, owner: str | None = None): """Apply `mutate(index)` to the entity index without losing a concurrent writer's entry: take the lock, **re-read the index fresh** (not a stale in-memory copy), run the mutator, save, release. Returns the freshly-read-and-saved index. The fresh re-read inside the critical section is the core of the fix — even if the advisory lock can't be taken (another session active), reading immediately before the save collapses the read-modify-write window to two calls instead of spanning the whole capture, so an unlocked update is still far safer than the prior code. """ with vault_lock(owner) as held: index = idx_mod.load() mutate(index) idx_mod.save(index) if not held: print("echo_concurrency: entity index updated WITHOUT the advisory lock " "(another session may be active); the fresh re-read minimizes the race.", file=sys.stderr) return index