jason/chorus
1
0
Fork 0
forked from jason/echo
Code Pull Requests Activity

Phase 4: shared-write hardening + group lenses

Browse Source 
- vault_lock gains bounded retry/backoff (CHORUS_LOCK_RETRIES=4,
  CHORUS_LOCK_BACKOFF=0.5s); atomic_index_update and recall-index upkeep
  retry acquisition — unlocked fallback only after exhausted retries,
  loudly, so an already-written note is never stranded unindexed.
- Duplicate-gate candidates carry the existing note's author; STOP
  message says whose entity the capture collided with.
- recall --author <member> filters hits + linked context to one member's
  notes; recall briefs and --json surface the author: stamp.
- load --all-members appends a group-wide view (every member's scope +
  last-session heartbeat); roster parsing accepts folders as
  trailing-slash file entries (the REST API's actual shape).
- operating-contract.md concurrency section rewritten for N members.
- Two-member e2e suite committed as eval/test_multimember.py (26 checks
  incl. all four Phase-4 behaviors).
- Manifest -> 2.0.0-alpha.4; rebuilt chorus-memory.plugin.

Verified: 25/25 unit, scaffold, routing-sync, 5 mock e2e suites (incl.
new multimember), run_eval metrics unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Jason Stedwell
2026-07-21 15:29:55 -05:00
parent 33a2d73a22
commit 2ddf199850
14 changed files with 302 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chorus-memory.plugin.src/skills/chorus-memory/scripts/chorus.py
+40 -6
View File
@@ -31,7 +31,7 @@ Config (group/member/endpoint/key are machine-local — see chorus_config; the p
CHORUS_TODAY YYYY-MM-DD to use as "today" (pass the conversation's currentDate)
Usage:
chorus.py load # cold-start: the 8 orientation reads in one call
chorus.py load [--all-members] # cold-start: the 8 orientation reads (+ optional group-wide view)
chorus.py get <path> # print file contents (404 -> exit 44)
chorus.py map <path> # document-map JSON (headings/blocks/frontmatter)
chorus.py ls <dir> # directory listing JSON
@@ -626,12 +626,14 @@ def cmd_scope(subcommand: str, text: str | None = None, as_json: bool = False) -
return 0
def cmd_load() -> int:
def cmd_load(all_members: bool = False) -> int:
"""Cold-start orientation: the canonical 8 reads in one call — the marker, the
GROUP profile, then MY member state (preferences/context/heartbeat/inbox), the
shared daily note, and the group inbox. 404s on today's daily note and the
inboxes are normal (printed as absent, not errors). If the vault is
bootstrapped but MY namespace is missing, self-onboard it."""
bootstrapped but MY namespace is missing, self-onboard it. With
`all_members`, append a compact group-wide view: every member's active scope
and last-session pointer ("what has the team been doing")."""
if not chorus_config.is_configured(_CFG):
print("chorus: NOT CONFIGURED — this machine has no usable CHORUS key file yet.")
print(f"Expected at: {chorus_config.config_path()}")
@@ -725,6 +727,34 @@ def cmd_load() -> int:
for f in files[:5]:
print(f" {member_sessions_dir()}/{f}")
print()
# Group-wide orientation: one compact line-pair per member (scope + last
# session). A lens over everyone's namespaces — read-only, opt-in.
if all_members and not offline:
st, body = request("GET", vault_url("_agent/members/"))
roster = []
if st == 200:
try:
payload = json.loads(body)
# Folders may arrive under "folders" or as trailing-slash entries
# in "files" (both shapes exist in the wild — cf. vault_lint.list_dir).
entries = list(payload.get("files", [])) + list(payload.get("folders", []))
roster = sorted({e[:-1] for e in entries if e.endswith("/")})
except json.JSONDecodeError:
roster = []
print(f"===== group: all members ({len(roster)}) =====")
for m in roster:
me = " (me)" if m == MEMBER else ""
st_c, b_c = request("GET", vault_url(member_path("current-context.md", m)))
scope = ""
if st_c == 200:
scope = " ".join(extract_heading(b_c.decode(errors="replace"), "Scope").split())
scope = re.sub(r"<!--.*?-->", "", scope).strip()[:140]
st_h, b_h = request("GET", vault_url(member_path("heartbeat.md", m)))
last = b_h.decode(errors="replace").strip()[:120] if st_h == 200 else "(no sessions yet)"
print(f" {m}{me}")
print(f" scope: {scope or '(not set)'}")
print(f" last: {last}")
print()
if offline:
print("NOTE: vault unreachable — context above is last-known-good cache; writes this "
"session will be queued and synced when the vault returns.")
@@ -780,7 +810,9 @@ def cmd_config(args) -> int:
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Validated cross-platform CHORUS vault client")
sub = parser.add_subparsers(dest="cmd", required=True)
sub.add_parser("load")
p = sub.add_parser("load")
p.add_argument("--all-members", action="store_true",
help="append a group-wide orientation: every member's scope + last session")
sub.add_parser("flush") # H2: replay writes queued during a prior outage
sub.add_parser("doctor") # M3: one-call readiness check
sub.add_parser("get").add_argument("path")
@@ -819,6 +851,7 @@ def build_parser() -> argparse.ArgumentParser:
# High-level operations (entity index + cross-links + recall). See chorus_ops.py.
sub.add_parser("resolve").add_argument("mention", nargs="+")
p = sub.add_parser("recall"); p.add_argument("query", nargs="+"); p.add_argument("--limit", type=int, default=6)
p.add_argument("--author") # filter hits to one member's notes (a lens, not a wall)
p.add_argument("--json", action="store_true") # structured hits + neighbourhood
p = sub.add_parser("link"); p.add_argument("a"); p.add_argument("b")
p.add_argument("--json", action="store_true")
@@ -845,7 +878,7 @@ def main(argv: list[str] | None = None) -> int:
args = build_parser().parse_args(argv)
try:
if args.cmd == "load":
return cmd_load()
return cmd_load(all_members=args.all_members)
if args.cmd == "doctor":
import chorus_doctor
return chorus_doctor.run()
@@ -918,7 +951,8 @@ def main(argv: list[str] | None = None) -> int:
if args.cmd == "resolve":
return chorus_ops.resolve(" ".join(args.mention))
if args.cmd == "recall":
return chorus_ops.recall(args.query, limit=args.limit, as_json=args.json)
return chorus_ops.recall(args.query, limit=args.limit, as_json=args.json,
author=args.author)
if args.cmd == "link":
return chorus_ops.link(args.a, args.b, as_json=args.json)
return chorus_ops.capture(
chorus-memory.plugin.src/skills/chorus-memory/scripts/chorus_concurrency.py
+24 -8
View File
@@ -23,12 +23,19 @@ from __future__ import annotations
import contextlib
import os
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import chorus # noqa: E402
import chorus_index as idx_mod # noqa: E402
# Bounded retry for index-write locks (Phase 4 hardening). With N members writing,
# lock holds are sub-second, so a short backoff nearly always wins the lock; the
# unlocked fallback (loud, fresh-re-read) survives only a genuinely stuck peer.
INDEX_LOCK_RETRIES = max(0, int(os.environ.get("CHORUS_LOCK_RETRIES", "4")))
INDEX_LOCK_BACKOFF = float(os.environ.get("CHORUS_LOCK_BACKOFF", "0.5"))
def auto_owner() -> str:
"""A stable-per-process owner id: member + client tag + pid, so lock contention
@@ -41,16 +48,24 @@ def auto_owner() -> str:
@contextlib.contextmanager
def vault_lock(owner: str | None = None, required: bool = False):
def vault_lock(owner: str | None = None, required: bool = False,
retries: int = 0, backoff: float = INDEX_LOCK_BACKOFF):
"""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
Yields `held: bool`. The lock is cooperative (read-back-confirmed, TTL-reclaimable).
`retries` > 0 re-attempts acquisition with linear backoff (attempt N sleeps
N*backoff) — with multiple members writing, lock holds are sub-second, so a short
retry almost always acquires. If it still can't: 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 = chorus.cmd_lock(owner, quiet=True)
attempt = 0
while rc != 0 and attempt < retries:
attempt += 1
time.sleep(backoff * attempt)
rc = chorus.cmd_lock(owner, quiet=True)
held = rc == 0
if not held and required:
raise chorus.ChorusError(f"vault_lock: could not acquire lock for '{owner}' (held elsewhere)", 75)
@@ -69,12 +84,13 @@ def atomic_index_update(mutate, owner: str | None = None):
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.
The fresh re-read inside the critical section is the core of the fix. Acquisition
retries with bounded backoff (INDEX_LOCK_RETRIES) — in a group vault the lock is
only ever held for sub-second index writes, so retries nearly always acquire; the
unlocked path (fresh re-read + loud warning) survives only a genuinely stuck peer,
chosen over hard failure so a note that was already written never goes unindexed.
"""
with vault_lock(owner) as held:
with vault_lock(owner, retries=INDEX_LOCK_RETRIES) as held:
index = idx_mod.load()
mutate(index)
idx_mod.save(index)
chorus-memory.plugin.src/skills/chorus-memory/scripts/chorus_ops.py
+21 -4
View File
@@ -73,11 +73,11 @@ def link(a_path: str, b_path: str, as_json: bool = False) -> int:
# ----------------------------------------------------------------- recall -----
def recall(query, limit: int = 8, as_json: bool = False) -> int:
def recall(query, limit: int = 8, as_json: bool = False, author: str | None = None) -> int:
"""Hybrid lexical (BM25) + graph recall — implemented in chorus_recall. Kept here as
the stable public entrypoint (chorus.py and /chorus-recall call chorus_ops.recall)."""
import chorus_recall # lazy: only pulls the BM25 modules when recall is actually run
return chorus_recall.recall(query, limit=limit, as_json=as_json)
return chorus_recall.recall(query, limit=limit, as_json=as_json, author=author)
# ----------------------------------------------------------- agent log -------
@@ -112,6 +112,21 @@ def ensure_daily_log(line: str) -> None:
# ----------------------------------------------------------------- capture ---
def _note_author(path: str) -> str:
"""Best-effort author: of an existing note — so gate messages can say WHOSE
entity a capture collided with. Never raises; empty string on any miss."""
if not path:
return ""
try:
status, body = chorus.request("GET", chorus.vault_url(path))
if status == 200:
m = re.search(r"(?m)^author:\s*(\S+)", body.decode(errors="replace")[:2000])
return m.group(1) if m else ""
except Exception: # noqa: BLE001
pass
return ""
def _build_note(fm_type: str, status_v: str, today_s: str, title: str,
body_text: str, sources: list[str], aliases: list[str] | None = None,
tags: list[str] | None = None) -> str:
@@ -245,7 +260,8 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
gate_hits = []
if not existing_reachable and not force and not dry_run:
gate_hits = [{"slug": s, "path": c.get("path"), "title": c.get("title"),
"kind": c.get("kind"), "score": sc}
"kind": c.get("kind"), "score": sc,
"author": _note_author(c.get("path", ""))}
for s, c, sc in idx_mod.gate_candidates(index, title, kind=kind,
threshold=DUP_GATE)]
if gate_hits:
@@ -259,7 +275,8 @@ def capture(kind: str | None, title: str, file_arg: str | None, status_v: str =
print(f"STOP: '{title}' likely already exists — not creating a duplicate.",
file=real_stdout)
for c in gate_hits:
print(f" candidate: {c['slug']} -> {c['path']} (score {c['score']})",
by = f", created by {c['author']}" if c.get("author") else ""
print(f" candidate: {c['slug']} -> {c['path']} (score {c['score']}{by})",
file=real_stdout)
print(" re-run with --merge-into <slug> to update the existing entity, "
"or --force to create anyway.", file=real_stdout)
chorus-memory.plugin.src/skills/chorus-memory/scripts/chorus_recall.py
+20 -2
View File
@@ -316,7 +316,7 @@ def update_note(path: str, text: str) -> None:
if not _indexable(path):
return # only entity notes are part of the recall corpus
import chorus_concurrency
with chorus_concurrency.vault_lock():
with chorus_concurrency.vault_lock(retries=chorus_concurrency.INDEX_LOCK_RETRIES):
ix = load_index() # fresh read inside the lock
ix.add(path, text) # raw text: add() strips + extracts meta
save_index(ix)
@@ -385,6 +385,9 @@ def _doc_summary(path: str, text: str) -> dict:
ms = _FM_STATUS.search(text[:2000])
if ms:
out["status"] = ms.group(1)
ma = re.search(r"(?m)^author:\s*(\S+)", text[:2000])
if ma:
out["author"] = ma.group(1)
body = strip_frontmatter(text)
status = re.search(r"(?ms)^##\s+Status\s*$(.*?)(?=^##\s|\Z)", body)
if status and status.group(1).strip():
@@ -414,6 +417,8 @@ def _brief(path: str, score: float | None = None, via: str | None = None) -> Non
stamps.append(f"updated: {info['updated']}")
if info.get("status"):
stamps.append(f"status: {info['status']}")
if info.get("author"):
stamps.append(f"author: {info['author']}")
if stamps:
print("_" + " · ".join(stamps) + "_") # staleness is visible, not guessed
if info.get("excerpt"):
@@ -421,7 +426,7 @@ def _brief(path: str, score: float | None = None, via: str | None = None) -> Non
# ------------------------------------------------------------------- entrypoint
def recall(query, limit: int = 8, as_json: bool = False) -> int:
def recall(query, limit: int = 8, as_json: bool = False, author: str | None = None) -> int:
q = " ".join(query) if isinstance(query, list) else query
today_s = chorus.today()
index = idx_mod.load()
@@ -464,6 +469,19 @@ def recall(query, limit: int = 8, as_json: bool = False) -> int:
# --- graph layer ----------------------------------------------------------
neighbours = expand_graph(hits, nmap, base, max_hops=MAX_HOPS)
# --- optional member filter (--author): keep only notes whose frontmatter
# author matches. Group memory is shared, so this is a lens, not a wall.
if author:
cand = hits + [p for p, _ in neighbours]
atexts = chorus.read_many(cand)
def _by(p: str) -> bool:
m = re.search(r"(?m)^author:\s*(\S+)", (atexts.get(p) or "")[:2000])
return bool(m) and m.group(1) == author
hits = [p for p in hits if _by(p)]
neighbours = [(p, sv) for p, sv in neighbours if _by(p)]
if as_json:
import chorus_output
texts = chorus.read_many(hits + [p for p, _ in neighbours[: 2 * limit]])