Files
multi-agent-mux/.agents/skills/lib_py/loop/store.py
T
GodopuandClaude Opus 5 01b465f50c fix(issue-5): bind the auto-completion decision atomically to its attempt
Two findings from round 2, both reproduced before being accepted.

R2 -- attribution was compared before the terminal write and nothing tied the
two together. A job could complete, be re-dispatched and claimed again in that
window, and evidence from the first attempt then satisfied a status check and
finished the second one. Reproduced with real transitions: committed, with the
registry already on iteration 2.

This corrects my own earlier reasoning, and I want to be plain about it. Last
round I argued that the attempt guard could live entirely in the evidence layer
because that is where attempt identity lives. That holds for comparing evidence
to a dispatch, and it does not hold for atomicity: a precondition can only be
atomic with the commit if the transaction can read it under its own lock. So the
identity now lives in the record.

It is an opaque dispatch token, not the iteration counter. That distinction is
the part of the earlier rebuttal that still stands: iteration already means the
loop round, and comparing a dispatch attempt against it produces a guard that is
vacuous when they agree and wrong when they diverge. A token has no competing
meaning. It is minted when an attempt begins -- claim and redispatch -- and the
gate passes the token it decided against, so any intervening re-dispatch refuses
the commit.

R4 -- the intent identity was being emitted but nothing consumed it. A replayed
notification is re-published with a fresh wire sequence, so the monotonic guard
cannot recognise it; to that guard a retry looks new. The subscriber now
deduplicates on the stable intent identity, leaving the sequence guard as the
rule for ordinary events. Across a restart the disk fallback resolves terminal
status and a duplicate terminal is refused as an idempotent ack, so a replay
after restart cannot double-apply either.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
2026-09-06 23:21:11 +09:00

114 lines
4.0 KiB
Python

"""Shadow-owned store for dispatch and attribution evidence.
Deliberately NOT the job record. Two constraints force this:
* phase 1 froze the registry record schema, and
* P2A-03 forbids this phase from writing the registry at all.
The record has none of `attempt_id`, `agent_session_uuid` or
`pane_process_generation`, so four-fold attribution has nowhere to live in it.
Keeping the evidence here means shadow mode can carry full attribution while the
registry stays byte-for-byte untouched. Whether these fields are promoted into
the record is a phase 2b decision this phase does not prejudge.
"""
from __future__ import annotations
import json
import os
import tempfile
from pathlib import Path
from typing import Any, Dict, Optional
SHADOW_DIRNAME = os.path.join(".mam", "shadow")
def shadow_root(workspace: Optional[str] = None) -> Path:
return Path(workspace or os.getcwd()) / SHADOW_DIRNAME
def attribution_path(job_id: str, workspace: Optional[str] = None) -> Path:
return shadow_root(workspace) / "attribution" / "{}.json".format(job_id)
def decisions_path(job_id: str, workspace: Optional[str] = None) -> Path:
return shadow_root(workspace) / "{}.ndjson".format(job_id)
def _atomic_write(path: Path, text: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".{}.".format(path.name), suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(text)
fh.flush()
os.fsync(fh.fileno())
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
os.unlink(tmp)
raise
def record_dispatch(job_id, attempt_id, agent_session_uuid, pane_process_generation,
dispatch_ack=True, workspace=None, dispatch_token=None) -> Dict[str, Any]:
"""Persist the four identifiers at dispatch time.
The execution anchor starts false on purpose: dispatch being acknowledged is
not evidence that the attempt ran.
"""
context = {
"job_id": job_id,
"attempt_id": attempt_id,
"agent_session_uuid": agent_session_uuid,
"pane_process_generation": pane_process_generation,
"dispatch_ack": bool(dispatch_ack),
"execution_anchor_observed": False,
# The record's token for the attempt this evidence belongs to. Carrying
# it here is what lets the gate prove the evidence and the commit are
# about the same attempt.
"dispatch_token": dispatch_token,
}
_atomic_write(attribution_path(job_id, workspace),
json.dumps(context, ensure_ascii=False, indent=2) + "\n")
return context
def mark_execution_anchor(job_id, workspace=None) -> Optional[Dict[str, Any]]:
"""Record that this attempt was actually observed executing after dispatch."""
context = load_dispatch(job_id, workspace)
if context is None:
return None
context["execution_anchor_observed"] = True
_atomic_write(attribution_path(job_id, workspace),
json.dumps(context, ensure_ascii=False, indent=2) + "\n")
return context
def load_dispatch(job_id, workspace=None) -> Optional[Dict[str, Any]]:
path = attribution_path(job_id, workspace)
if not path.exists():
return None
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
return None
def append_decision(job_id, entry: Dict[str, Any], workspace=None) -> None:
"""Append one would-complete decision. The only thing shadow mode emits."""
path = decisions_path(job_id, workspace)
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "a", encoding="utf-8") as fh:
fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
def read_decisions(job_id, workspace=None) -> list:
path = decisions_path(job_id, workspace)
if not path.exists():
return []
out = []
for line in path.read_text(encoding="utf-8").splitlines():
if line.strip():
out.append(json.loads(line))
return out