From 568385c77a4e95510b6e2f69a714116b444467a3 Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 30 Oct 2026 05:23:51 +0900 Subject: [PATCH] fix(reset): fence every job registry the reset reads, by its physical directory (#78) Review round 3 R1 (codex): the registration guard chose the workspace marker from the path's spelling, so a symlink alias of .mam/jobs or a MAM_JOBS_DIR registry was not refused. The reset now writes .reset-in-progress into each registry its open-job check reads (.mam/jobs, MAM_JOBS_DIR, a row's other control_plane.mam_jobs_dir - the latter only for this workspace's sessions), each under that registry's lock and before the check, and removes it on exit; register_job reads the marker inside the registry it writes to. The precheck's separate lock barrier is subsumed by the fence. Co-Authored-By: Claude Opus 5.5 --- .agents/skills/lib_py/delegation/registry.py | 38 ++--- .agents/skills/lib_py/reset.py | 54 ++++++- .../multi-agent-mux-delegate-job/registry.md | 4 +- .agents/skills/multi-agent-mux-reset/SKILL.md | 4 +- .../multi-agent-mux-reset/scripts/reset.sh | 25 +++- tests/test_issue78_reset.py | 136 ++++++++++++------ 6 files changed, 186 insertions(+), 75 deletions(-) diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 1722fb9..22a5374 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -68,12 +68,16 @@ def liveness_window(heartbeat_sec: int, heartbeat_misses: int) -> int: # Issue #78: a reset of a workspace purges its agents' conversations, so no job -# may be registered in that workspace's .mam/jobs while one runs. The reset -# publishes .mam/reset-in-progress (pid/lstart, loop_lock.sh's format) and then -# takes this registry's lock once before it lists the open jobs; the marker is -# read under the same lock here, so a registration either is written before -# that listing or sees the marker. The reset's own create (the onboarding job -# of an agent it recreates) passes the reset's pid in MAM_RESET_OWNER_PID. +# may be registered for them while it runs. The reset writes RESET_MARKER into +# every registry its open-job check reads (its own .mam/jobs and MAM_JOBS_DIR, +# and a row's other control_plane.mam_jobs_dir), each under that registry's +# lock and before the check: a registration reads it under the same lock, so it +# is either written before the check or refused - by whatever path it names +# the directory. The marker carries the reset's pid/lstart (loop_lock.sh's +# format), its workspace, and for another control plane's registry the +# sessions it covers. The reset's own create (the onboarding job of an agent +# it recreates) passes the reset's pid in MAM_RESET_OWNER_PID. +RESET_MARKER = ".reset-in-progress" RESET_OWNER_ENV = "MAM_RESET_OWNER_PID" @@ -90,15 +94,13 @@ def _lstart(pid: str) -> str: return " ".join(out.split()) -def reset_refusal(registry_dir: str) -> Optional[str]: - """Why a registration in ``registry_dir`` is refused, or None: a live - reset of the workspace that owns it (``/.mam/jobs``) that is not the - caller's own. Holder states as loop_lock.sh decides them: a dead or - reused pid is stale; ps failing on this process is unknown (refused).""" - jobs = os.path.abspath(registry_dir) - if os.path.basename(jobs) != "jobs" or os.path.basename(os.path.dirname(jobs)) != ".mam": - return None - marker = os.path.join(os.path.dirname(jobs), "reset-in-progress") +def reset_refusal(registry_dir: str, agent_session: str) -> Optional[str]: + """Why registering a job for ``agent_session`` in ``registry_dir`` is + refused, or None: a live reset fenced this registry (for that session) and + the caller is not its own create. Holder states as loop_lock.sh decides + them: a dead or reused pid is stale; ps failing on this process is unknown + (refused).""" + marker = os.path.join(registry_dir, RESET_MARKER) try: with open(marker, encoding="utf-8") as fh: fields = dict(line.split("=", 1) for line in fh.read().splitlines() if "=" in line) @@ -106,6 +108,8 @@ def reset_refusal(registry_dir: str) -> Optional[str]: return None except (OSError, ValueError) as exc: return "the reset marker %s cannot be read (%s)" % (marker, exc) + if "sessions" in fields and agent_session not in fields["sessions"].split(","): + return None pid = fields.get("pid", "") if not pid.isdigit() or os.environ.get(RESET_OWNER_ENV) == pid: return None @@ -115,7 +119,7 @@ def reset_refusal(registry_dir: str) -> Optional[str]: if not cur or (fields.get("lstart") and cur != fields["lstart"]): return None return ("a reset of %s is in progress (pid %s); its agents are being purged and recreated;" - " register the job when it has finished" % (os.path.dirname(os.path.dirname(jobs)), pid)) + " register the job when it has finished" % (fields.get("workspace", "?"), pid)) def register_job( @@ -213,7 +217,7 @@ def register_job( if dry_run: record["dry_run"] = True with registry_lock(registry_dir): - refusal = reset_refusal(registry_dir) + refusal = reset_refusal(registry_dir, agent_session) if refusal: raise ResetInProgressError(refusal) if mqtt_common._job_path(job_id, registry_dir).exists(): diff --git a/.agents/skills/lib_py/reset.py b/.agents/skills/lib_py/reset.py index b2a150e..49adb16 100644 --- a/.agents/skills/lib_py/reset.py +++ b/.agents/skills/lib_py/reset.py @@ -18,11 +18,12 @@ import sqlite3 import subprocess import sys import tempfile -from typing import Dict, List, Mapping, Optional, Tuple +from typing import Dict, Iterator, List, Mapping, Optional, Tuple from lib_py.agents.base import AGENT_SESSION_ENV, POST_DISCOVERED from lib_py.agents.registry import agent_of_row, get_adapter, own_key from lib_py.delegation.mqtt_common import registry_lock +from lib_py.delegation.registry import RESET_MARKER from lib_py.update import open_jobs, row_server from lib_py.verify_session import verify_session_uuid @@ -484,6 +485,39 @@ def target_open_jobs(workspace: str, state: dict, env: Mapping[str, str], sessio return out +def fence_registries(workspace: str, state: dict, env: Mapping[str, str]) -> Iterator[str]: + """Write RESET_MARKER into every registry open_jobs reads, each under its + registry lock (a registration that read no marker has finished writing), + and yield its real path as it is written. This workspace's .mam/jobs and + MAM_JOBS_DIR (created when absent) refuse every registration; another + control plane's registry named by a row refuses only those addressed to + this registry's rows. The marker copies .mam/reset-in-progress.""" + with open(os.path.join(workspace, ".mam", "reset-in-progress"), encoding="utf-8") as fh: + base = "".join(l + "\n" for l in fh.read().splitlines() if l.split("=", 1)[0] in ("pid", "lstart", "started")) + base += "workspace=%s\n" % workspace + rows = state.get("herdr_sessions") or [] + sessions = ",".join("herdr:%s" % s["name"] for s in rows if s.get("name")) + plan, seen = [], set() + for d in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or ""): + if d: + os.makedirs(d, exist_ok=True) + plan.append((d, base)) + for s in rows: + d = (s.get("control_plane") or {}).get("mam_jobs_dir") or "" + if d and os.path.isdir(d): + plan.append((d, base + "sessions=%s\n" % sessions)) + for d, text in plan: + if _real(d) in seen: + continue + seen.add(_real(d)) + with registry_lock(d): + fd, tmp = tempfile.mkstemp(dir=d, prefix=RESET_MARKER + ".") + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(text) + os.replace(tmp, os.path.join(d, RESET_MARKER)) + yield _real(d) + + def target_id_owned(state: dict, session_name: str) -> bool: """The row's recorded post-discovered id still verifies (revalidate) as a conversation of the row's cwd - the check stop makes before it keeps such @@ -757,6 +791,9 @@ def main(argv: Optional[List[str]] = None) -> int: s.add_argument("--workspace", required=True) s.add_argument("--session", required=True) + s = sub.add_parser("fence") + s.add_argument("--workspace", required=True) + s = sub.add_parser("id-owned") s.add_argument("--workspace", required=True) s.add_argument("--session", required=True) @@ -814,12 +851,6 @@ def main(argv: Optional[List[str]] = None) -> int: return 1 if sel["errors"] else 0 if a.cmd == "precheck": - # A registration that read no reset marker finishes writing under this - # lock before the open jobs are listed (registry.reset_refusal). - jobs_dir = os.path.join(ws, ".mam", "jobs") - if os.path.isdir(jobs_dir): - with registry_lock(jobs_dir): - pass lines = precheck(ws, state, os.environ) for line in lines: print(line) @@ -852,6 +883,15 @@ def main(argv: Optional[List[str]] = None) -> int: print(line) return 1 if lines else 0 + if a.cmd == "fence": + try: + for d in fence_registries(ws, state, os.environ): + print(d, flush=True) + except OSError as exc: + print("ERROR: cannot fence the job registries (%s)" % exc, file=sys.stderr) + return 1 + return 0 + if a.cmd == "id-owned": return 0 if target_id_owned(state, a.session) else 1 diff --git a/.agents/skills/multi-agent-mux-delegate-job/registry.md b/.agents/skills/multi-agent-mux-delegate-job/registry.md index 0d70570..587a243 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/registry.md +++ b/.agents/skills/multi-agent-mux-delegate-job/registry.md @@ -141,8 +141,8 @@ $PY scripts/registry.py status --job --set completed # set status $PY scripts/registry.py pick --agent-session herdr:claude # claim → running ``` -Exit codes: `0` ok, `1` not found / bad status / (`register`) a reset of the -workspace owning `/.mam/jobs` is in progress (Issue #78), `3` (`pick`) no +Exit codes: `0` ok, `1` not found / bad status / (`register`) a reset fenced +this registry (its `.reset-in-progress` names a live reset, Issue #78), `3` (`pick`) no pending job for that session. --- diff --git a/.agents/skills/multi-agent-mux-reset/SKILL.md b/.agents/skills/multi-agent-mux-reset/SKILL.md index cb2fa29..b400073 100644 --- a/.agents/skills/multi-agent-mux-reset/SKILL.md +++ b/.agents/skills/multi-agent-mux-reset/SKILL.md @@ -23,7 +23,7 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run] | Targets | Every registry row with `status: running` and `realpath(pane.cwd) == realpath(WS)`. Rows with any other status, worktree rows, and `orchestrator_uuids` are skipped and listed with their reason. | | Confirmation | `[y/N]` prompt; without a terminal `-y`/`--yes` is required (exits `3` otherwise). `--dry-run` inspects without `--yes` and changes nothing. | | Prechecks | Refuses before any change when an open job exists (prints the `registry.py ... close --request cancelled` command), when `.mam/loop-guard-active` or `.mam/update-in-progress` is held, when a target row lacks a valid agent or role, when a running `claude`/`grok` target has no recorded conversation id (the monitor had not pinned one; record it, or stop/purge it by hand first), when `.mam/reset-spec.json` is invalid, or when the caller itself is one of the targets. | -| Lock & concurrency | Takes the update admission lock exclusively, acquires `.mam/loop-guard-active` and publishes `.mam/reset-in-progress`, then releases the admission lock and runs the prechecks. `run_loop.sh`, `update.sh` and another reset refuse on the loop marker; every job registration in this workspace's `.mam/jobs` (`multi-agent-mux-delegate-job submit`, `registry.py register`, the registry API) refuses on the reset marker, except the onboarding jobs of the reset's own `create_session.sh`. | +| Lock & concurrency | Takes the update admission lock exclusively, acquires `.mam/loop-guard-active` and publishes `.mam/reset-in-progress`, then releases the admission lock and runs the prechecks. `run_loop.sh`, `update.sh` and another reset refuse on the loop marker. The reset then writes `.reset-in-progress` into every job registry its open-job check reads (`.mam/jobs`, `MAM_JOBS_DIR`, and a row's other `control_plane.mam_jobs_dir`), each under that registry's lock, before the check. A registration (`multi-agent-mux-delegate-job submit`, `registry.py register`, the registry API) into a fenced registry - by any path to that directory - refuses; in another control plane's registry only for this workspace's sessions. The onboarding jobs of the reset's own `create_session.sh` are let through. | | Spec file & rerun | Writes `.mam/reset-spec.json` atomically before the first stop/purge and removes it only when all targets reach `done`. Rerunning resumes from the spec file without re-purging already recreated agents (`created` or `done`). | | Sequence per target | Sequential (never parallel): ⓪ re-checks that its row still has the spec's agent and herdr server and that no job addressed to it is open, ① normal `stop_session.sh --reason reset` (a post-discovered agent's id is captured onto the row while herdr is live; `status: stopped`), ② checks that the row's id is the one recorded before the stop and, for a post-discovered agent, that it still verifies as a conversation of the row's folder, ③ `stop_session.sh --reason reset --purge-conversation --yes` (using the row's recorded ID, never a CLI UUID), then `create_session.sh` with the original name, agent, role, `--herdr-session`, and `--herdr-workspace` (onboarding enabled by default). | | When skipped (exit 4) | The target is skipped and the reset continues to the next agent when step ⓪ finds a changed row (`row changed`) or an open job (`job open`) - nothing was changed for it -, or when step ② finds no id (`no conversation id recorded`) or an id other than the recorded one or one that does not verify for the row's folder (`id not verified`), or step ③ exits `4` (`purge refused`, in-use gate) - it is stopped but resumable (its row keeps its id) and its conversation files are untouched. Rerun once the cause is gone, or resume it. | @@ -44,5 +44,5 @@ bash .agents/skills/multi-agent-mux-reset/scripts/reset.sh --yes [--dry-run] - **No external process termination**: processes holding conversation files open (reported via `lsof` in `in_use_hints`) are never signalled by MAM; end them yourself when safe, then rerun. - **Shared stores preserved**: Grok's shared `memory-v2.db` is left intact during purge; only session rows and per-session files are removed. - **Orchestrator hook during reset**: while `.mam/loop-guard-active` is held by the reset, the AGY orchestrator's direct file-edit hook (`loop_delegation_guard.sh`) blocks direct edits until the reset finishes. -- **Job submissions during a reset**: `multi-agent-mux-delegate-job submit` and `registry.py register` refuse with `a reset of is in progress` until the reset ends; submit again afterwards. A record written into `.mam/jobs` by hand for a target after the precheck makes that target `job open` (skipped, not cancelled) if it is there before that target's stop. +- **Job submissions during a reset**: `multi-agent-mux-delegate-job submit` and `registry.py register` refuse with `a reset of is in progress` until the reset ends (a marker left by a killed reset is stale and ignored); submit again afterwards. A record written into `.mam/jobs` by hand for a target after the precheck makes that target `job open` (skipped, not cancelled) if it is there before that target's stop. - **Same-type post-discovered ordering**: when resetting multiple post-discovered agents of the same type, the second `create_session.sh` waits up to `MAM_ID_ORDER_WAIT_S` for the first agent's ID to be recorded; if it times out, rerun the reset once the ID is recorded. diff --git a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh index 7d3afef..31600ab 100755 --- a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh +++ b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh @@ -92,9 +92,8 @@ _refuse_locked() { # The update admission lock is taken exclusively, as update.sh does, while the # reset publishes its markers: a delegate submit admitted earlier has written # its job record (the precheck below sees it), and one admitted later sees -# .mam/reset-in-progress and refuses. Any other registration in .mam/jobs -# (registry.py register, the delegation API) reads the marker under the -# registry lock, which the precheck takes before it lists the open jobs. +# .mam/reset-in-progress and refuses. Any other registration (registry.py +# register, the delegation API) is fenced below. mkdir -p "$MAM" exec 7>>"$MAM/update-admission.lock" case "$("$PY" -m lib_py.pane_lock acquire --fd 7 --wait-env MAM_UPDATE_ADMISSION_WAIT_S)" in @@ -115,7 +114,12 @@ case "$_acquire_rc" in 3) _refuse_locked "a review loop or another reset is running (pid $(_mam_lock_field "$MARKER" "pid" 2>/dev/null || echo "unknown")); wait for it to end" ;; *) _refuse_locked "cannot determine whether a review loop or another reset is running (ps unavailable?)" ;; esac +FENCED=() _release_reset_lock() { + local d + for d in ${FENCED[@]+"${FENCED[@]}"}; do + [ "$(_mam_lock_field "$d/.reset-in-progress" pid 2>/dev/null || true)" != "$$" ] || rm -f "$d/.reset-in-progress" + done mam_release_loop_lock "$RESET_MARKER" || true mam_release_loop_lock "$MARKER" || true } @@ -140,6 +144,21 @@ if ! STATE_JSON="$(load_state_json strict)"; then exit 1 fi +# Every registry the precheck reads gets the reset marker under its own lock +# (lib_py.reset fence): a registration that read none is written by then, any +# later one is refused, by whatever path it names that registry. +fence_out="" +fence_rc=0 +fence_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset fence --workspace "$WS")" || fence_rc=$? +while IFS= read -r line; do + [ -n "$line" ] && FENCED+=("$line") +done <<< "$fence_out" +if [ "$fence_rc" -ne 0 ]; then + echo "ERROR: reset refused; nothing was changed:" >&2 + echo " - the job registries could not be fenced against new registrations (lib_py.reset fence rc $fence_rc)" >&2 + exit 1 +fi + pre_out="" pre_rc=0 pre_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset precheck --workspace "$WS")" || pre_rc=$? diff --git a/tests/test_issue78_reset.py b/tests/test_issue78_reset.py index 27f5f9b..639f62c 100644 --- a/tests/test_issue78_reset.py +++ b/tests/test_issue78_reset.py @@ -1655,7 +1655,7 @@ if args[:2] == ["pane", "close"] or args[:1] == ["kill-session"]: "--agent-session", "herdr:{name_b}", "--prompt", "hi"], cwd={repr(str(mam_sandbox))}, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=120, ) - jobs = sorted(os.listdir({repr(str(jobs_dir))})) if os.path.isdir({repr(str(jobs_dir))}) else [] + jobs = sorted(n for n in os.listdir({repr(str(jobs_dir))}) if n.endswith(".json")) if os.path.isdir({repr(str(jobs_dir))}) else [] with open(probe, "w", encoding="utf-8") as fh: json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr, "marker": os.path.isfile(marker), "jobs": jobs}}, fh) @@ -1790,17 +1790,32 @@ def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path): assert res_stale.returncode == 0, res_stale.stderr -def test_registry_cli_is_refused_after_the_target_recheck(mam_sandbox: Path, mock_herdr: Path, mock_agents): - """Review round 2 R1 (codex): the documented registry CLI, run by a caller +@pytest.mark.parametrize("registry", ["canonical", "alias", "custom"]) +def test_registry_cli_is_refused_after_the_target_recheck( + mam_sandbox: Path, mock_herdr: Path, mock_agents, registry: str +): + """Review rounds 2-3 R1 (codex): the documented registry CLI, run by a caller that is not a child of the reset, after the target's open-job re-check and - before its stop, is refused by .mam/reset-in-progress and registers nothing; - the reset's own create still registers its onboarding job.""" + before its stop, is refused and registers nothing - whether it names the + registry by its canonical path, by a symlink alias, or the registry is the + MAM_JOBS_DIR the reset runs with; the reset's own create still registers + its onboarding job.""" import threading name = "reviewer-claude-01" _claude_transcript(mam_sandbox, mam_sandbox, UA) _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude") _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)]) + jobs_dir = mam_sandbox / ".mam" / "jobs" + jobs_dir.mkdir(parents=True, exist_ok=True) + env_extra = None + cli_dir = jobs_dir + if registry == "alias": + cli_dir = mam_sandbox / "job-alias" + cli_dir.symlink_to(jobs_dir) + elif registry == "custom": + jobs_dir = cli_dir = mam_sandbox / "elsewhere" / "jobs" + env_extra = {"MAM_JOBS_DIR": str(jobs_dir)} stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh" original = stop.with_name("stop-original.sh") stop.rename(original) @@ -1821,7 +1836,6 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]]) ) stop.write_text("#!/bin/bash\nexec %s %s \"$@\"\n" % (shlex.quote(sys.executable), shlex.quote(str(gate)))) stop.chmod(0o755) - jobs_dir = mam_sandbox / ".mam" / "jobs" record: dict = {} def writer(): @@ -1831,9 +1845,9 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]]) time.sleep(0.02) record["marker"] = (mam_sandbox / ".mam" / "reset-in-progress").is_file() before = sorted(p.name for p in jobs_dir.glob("*.json")) - registry = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py" + cli = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py" res = subprocess.run( - [sys.executable, str(registry), "--registry-dir", str(jobs_dir), "register", + [sys.executable, str(cli), "--registry-dir", str(cli_dir), "register", "--prompt", "external concurrent work", "--agent", "claude-code", "--agent-session", "herdr:" + name], capture_output=True, text=True, timeout=30, ) @@ -1845,7 +1859,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]]) worker = threading.Thread(target=writer) worker.start() try: - res = _run_reset(mam_sandbox, "--yes") + res = _run_reset(mam_sandbox, "--yes", env_extra=env_extra) finally: worker.join(timeout=90) _wait_reconcile_idle(mam_sandbox) @@ -1858,63 +1872,97 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]]) assert _result_of(res.stdout, name) == "done" onboarding = [json.loads(p.read_text(encoding="utf-8")) for p in jobs_dir.glob("*.json")] assert [j["agent_session"] for j in onboarding] == ["herdr:" + name], onboarding + assert not (jobs_dir / ".reset-in-progress").exists() + + +def _registry_marker(jobs_dir: Path, holder: subprocess.Popen, ws: Path, sessions: Optional[str] = None) -> None: + _live_marker(jobs_dir / ".reset-in-progress", holder) + with open(jobs_dir / ".reset-in-progress", "a", encoding="utf-8") as fh: + fh.write("workspace=%s\n" % ws) + if sessions is not None: + fh.write("sessions=%s\n" % sessions) def test_register_job_refuses_only_a_live_reset_not_its_own(mam_sandbox: Path, monkeypatch): from lib_py.delegation import registry - jobs_dir = str(mam_sandbox / ".mam" / "jobs") - marker = mam_sandbox / ".mam" / "reset-in-progress" + jobs_dir = mam_sandbox / ".mam" / "jobs" + other = mam_sandbox / "other-control-plane" / "jobs" + jobs_dir.mkdir(parents=True, exist_ok=True) + other.mkdir(parents=True) monkeypatch.delenv(registry.RESET_OWNER_ENV, raising=False) holder = subprocess.Popen(["sleep", "120"]) try: - _live_marker(marker, holder) - with pytest.raises(registry.ResetInProgressError, match="a reset of %s is in progress" % mam_sandbox): - registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=jobs_dir) - assert not list(Path(jobs_dir).glob("*.json")) - # Another control plane's registry is not this workspace's .mam/jobs - assert registry.reset_refusal(str(mam_sandbox / "elsewhere")) is None + _registry_marker(jobs_dir, holder, mam_sandbox) + _registry_marker(other, holder, mam_sandbox, sessions="herdr:a,herdr:b") + for d in (jobs_dir, other): + with pytest.raises(registry.ResetInProgressError, match="a reset of %s is in progress" % mam_sandbox): + registry.register_job(prompt="p", agent_session="herdr:b", registry_dir=str(d)) + # Another control plane's registry refuses only the reset's sessions + assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(other)) + assert not list(jobs_dir.glob("*.json")) monkeypatch.setenv(registry.RESET_OWNER_ENV, str(holder.pid)) - assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=jobs_dir) + assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(jobs_dir)) monkeypatch.delenv(registry.RESET_OWNER_ENV) finally: holder.kill() holder.wait() - assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=jobs_dir) - assert len(list(Path(jobs_dir).glob("*.json"))) == 2 + assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(jobs_dir)) + assert len(list(jobs_dir.glob("*.json"))) == 2 -def test_precheck_waits_for_a_registration_holding_the_registry_lock(tmp_path: Path): - """R1 lock order: a registration that read no reset marker holds the registry - lock until its record is written; the precheck, run after the marker is - published, takes that lock before it lists the open jobs, so it sees the job.""" +def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path: Path): + """R1 lock order and scope: a registration that read no marker holds the + registry lock until its record is written; fence writes the marker under + that lock, so it waits, and the precheck after it sees the job. Fenced are + .mam/jobs, MAM_JOBS_DIR (one marker for an alias of the same directory) and + a row's other control-plane registry (for this registry's sessions only).""" import fcntl ws = tmp_path / "ws" jobs_dir = ws / ".mam" / "jobs" jobs_dir.mkdir(parents=True) - state = {"herdr_sessions": [_row("reviewer-claude-01", "claude", ws, own_id=UA)]} - env = dict(os.environ, PYTHONPATH=str(SKILLS), PYTHONDONTWRITEBYTECODE="1", MAM_STATE_JSON=json.dumps(state)) - env.pop("MAM_JOBS_DIR", None) - lock = open(jobs_dir / ".lock", "a") - fcntl.flock(lock, fcntl.LOCK_EX) + (ws / "alias").symlink_to(jobs_dir) + other = tmp_path / "cp" / "jobs" + other.mkdir(parents=True) + row = _row("reviewer-claude-01", "claude", ws, own_id=UA) + row["control_plane"] = {"mam_jobs_dir": str(other)} + state = {"herdr_sessions": [row]} + env = dict(os.environ, PYTHONPATH=str(SKILLS), PYTHONDONTWRITEBYTECODE="1", MAM_STATE_JSON=json.dumps(state), + MAM_JOBS_DIR=str(ws / "alias")) + holder = subprocess.Popen(["sleep", "120"]) try: - proc = subprocess.Popen( - [sys.executable, "-m", "lib_py.reset", "precheck", "--workspace", str(ws)], - env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, - ) - time.sleep(1.5) - assert proc.poll() is None, proc.communicate() - (jobs_dir / "job78locked.json").write_text( - json.dumps({"job_id": "job78locked", "agent_session": "herdr:reviewer-claude-01", "status": "pending"}), - encoding="utf-8", - ) + _live_marker(ws / ".mam" / "reset-in-progress", holder) + lock = open(jobs_dir / ".lock", "a") + fcntl.flock(lock, fcntl.LOCK_EX) + try: + proc = subprocess.Popen( + [sys.executable, "-m", "lib_py.reset", "fence", "--workspace", str(ws)], + env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, + ) + time.sleep(1.5) + assert proc.poll() is None, proc.communicate() + assert not (jobs_dir / ".reset-in-progress").exists() + (jobs_dir / "job78locked.json").write_text( + json.dumps({"job_id": "job78locked", "agent_session": "herdr:reviewer-claude-01", "status": "pending"}), + encoding="utf-8", + ) + finally: + fcntl.flock(lock, fcntl.LOCK_UN) + lock.close() + out, err = proc.communicate(timeout=30) + assert proc.returncode == 0, out + err + assert out.split() == [os.path.realpath(jobs_dir), os.path.realpath(other)] + own = (jobs_dir / ".reset-in-progress").read_text(encoding="utf-8") + assert "pid=%d\n" % holder.pid in own and "workspace=%s\n" % os.path.realpath(ws) in own + assert "sessions=" not in own + assert "sessions=herdr:reviewer-claude-01\n" in (other / ".reset-in-progress").read_text(encoding="utf-8") + pre = subprocess.run([sys.executable, "-m", "lib_py.reset", "precheck", "--workspace", str(ws)], + env=env, capture_output=True, text=True, timeout=30) + assert pre.returncode == 1 and "job job78locked" in pre.stdout, pre.stdout + pre.stderr finally: - fcntl.flock(lock, fcntl.LOCK_UN) - lock.close() - out, err = proc.communicate(timeout=30) - assert proc.returncode == 1, out + err - assert "job job78locked" in out + holder.kill() + holder.wait() @pytest.mark.parametrize("artifact_cwd", ["own", "foreign"])