echo/echo-memory.plugin.src/skills/echo-memory/scripts/echo_concurrency.py

#!/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