fix(reset): never take another live reset's fence, and fence absent row registries (#78)

Review round 4 (codex):
- R1: fence_registries reads an existing marker under the registry lock and
  refuses the reset, changing nothing, when another live (or undecidable)
  reset holds it - registries may be shared between workspaces; only a stale
  marker is replaced, and the reset's exit removes only its own fences.
- R2: the registries to fence come from update.jobs_dirs, the enumeration
  open_jobs now uses too, whether or not they exist; an absent row registry
  is created and fenced (a registration would create it), and one that cannot
  be created refuses the reset.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_issue78_reset.py

@@ -1790,7 +1790,7 @@ def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path):
     assert res_stale.returncode == 0, res_stale.stderr
 
 
-@pytest.mark.parametrize("registry", ["canonical", "alias", "custom"])
+@pytest.mark.parametrize("registry", ["canonical", "alias", "custom", "row-absent"])
 def test_registry_cli_is_refused_after_the_target_recheck(
     mam_sandbox: Path, mock_herdr: Path, mock_agents, registry: str
 ):
@@ -1798,14 +1798,15 @@ def test_registry_cli_is_refused_after_the_target_recheck(
     that is not a child of the reset, after the target's open-job re-check and
     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."""
+    MAM_JOBS_DIR the reset runs with, or a row's control-plane registry that
+    did not exist when the reset started (round 4 R2); 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)])
+    row = _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
@@ -1816,6 +1817,10 @@ def test_registry_cli_is_refused_after_the_target_recheck(
     elif registry == "custom":
         jobs_dir = cli_dir = mam_sandbox / "elsewhere" / "jobs"
         env_extra = {"MAM_JOBS_DIR": str(jobs_dir)}
+    elif registry == "row-absent":
+        cli_dir = mam_sandbox / "elsewhere" / "jobs"
+        row["control_plane"] = {"mam_jobs_dir": str(cli_dir)}
+    _write_rows(mam_sandbox, [row])
     stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh"
     original = stop.with_name("stop-original.sh")
     stop.rename(original)
@@ -1844,7 +1849,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
             while not entered.exists() and time.monotonic() < end:
                 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"))
+            before = sorted(p.name for p in cli_dir.glob("*.json"))
             cli = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py"
             res = subprocess.run(
                 [sys.executable, str(cli), "--registry-dir", str(cli_dir), "register",
@@ -1852,7 +1857,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
                 capture_output=True, text=True, timeout=30,
             )
             record.update(rc=res.returncode, stdout=res.stdout, stderr=res.stderr, before=before,
-                          after=sorted(p.name for p in jobs_dir.glob("*.json")))
+                          after=sorted(p.name for p in cli_dir.glob("*.json")))
         finally:
             release.touch()
 
@@ -1873,6 +1878,7 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
     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()
+    assert not (cli_dir / ".reset-in-progress").exists()
 
 
 def _registry_marker(jobs_dir: Path, holder: subprocess.Popen, ws: Path, sessions: Optional[str] = None) -> None:
@@ -1916,7 +1922,8 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
     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)."""
+    a row's other control-plane registry (for this registry's sessions only),
+    also one that does not exist yet (created); a stale marker is replaced."""
     import fcntl
 
     ws = tmp_path / "ws"
@@ -1924,10 +1931,15 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
     jobs_dir.mkdir(parents=True)
     (ws / "alias").symlink_to(jobs_dir)
     other = tmp_path / "cp" / "jobs"
+    absent = tmp_path / "cp2" / "jobs"
     other.mkdir(parents=True)
+    # A stale marker (a killed reset) is replaced
+    (other / ".reset-in-progress").write_text("pid=999999\nlstart=x\nworkspace=/gone\n", encoding="utf-8")
     row = _row("reviewer-claude-01", "claude", ws, own_id=UA)
     row["control_plane"] = {"mam_jobs_dir": str(other)}
-    state = {"herdr_sessions": [row]}
+    row2 = _row("reviewer-claude-02", "claude", ws, own_id=UB)
+    row2["control_plane"] = {"mam_jobs_dir": str(absent)}
+    state = {"herdr_sessions": [row, row2]}
     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"])
@@ -1952,11 +1964,14 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
             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)]
+        assert out.splitlines() == ["fenced " + os.path.realpath(p) for p in (jobs_dir, other, absent)]
         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")
+        for d in (other, absent):
+            text = (d / ".reset-in-progress").read_text(encoding="utf-8")
+            assert "pid=%d\n" % holder.pid in text
+            assert "sessions=herdr:reviewer-claude-01,herdr:reviewer-claude-02\n" in text
         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
@@ -2010,3 +2025,100 @@ def test_post_discovered_id_is_purged_only_when_it_verifies_for_its_folder(
         kept = _read_rows(mam_sandbox)[name]
         assert kept["status"] == "stopped" and kept["codex_session_id_own"] == UB
         assert (mam_sandbox / ".mam" / "reset-spec.json").is_file()
+
+
+@pytest.mark.parametrize("other_has_target", [False, True])
+def test_other_reset_cannot_take_a_live_fence(
+    mam_sandbox: Path, mock_herdr: Path, mock_agents, other_has_target: bool
+):
+    """Review round 4 R1 (codex): while reset A holds the fence of a registry
+    shared with workspace B (both MAM_JOBS_DIR), B's reset is refused before
+    any change and leaves A's fence as it is; after B has ended, a registration
+    for A's target is still refused and A completes."""
+    import shutil
+    import threading
+
+    name = "reviewer-claude-01"
+    shared = mam_sandbox / "shared" / "jobs"
+    _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)])
+    other = mam_sandbox / "other-workspace"
+    shutil.copytree(SKILLS, other / ".agents" / "skills")
+    (other / ".mam").mkdir()
+    other_rows = []
+    other_tf = None
+    if other_has_target:
+        other_tf = _claude_transcript(mam_sandbox, other, UB)
+        other_rows = [_row("reviewer-claude-09", "claude", other, own_id=UB)]
+    (other / ".mam" / "agent-sessions.yaml").write_text(
+        yaml.safe_dump({"herdr_sessions": other_rows, "delegation_jobs": []}), encoding="utf-8")
+
+    stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh"
+    original = stop.with_name("stop-original.sh")
+    stop.rename(original)
+    entered, release = mam_sandbox / "stop-entered", mam_sandbox / "stop-release"
+    gate = mam_sandbox / "stop-gate.py"
+    gate.write_text(
+        f"""import os, pathlib, sys, time
+entered = pathlib.Path({str(entered)!r})
+release = pathlib.Path({str(release)!r})
+if not entered.exists():
+    entered.touch()
+    end = time.monotonic() + 90
+    while not release.exists() and time.monotonic() < end:
+        time.sleep(0.02)
+os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
+""",
+        encoding="utf-8",
+    )
+    stop.write_text("#!/bin/bash\nexec %s %s \"$@\"\n" % (shlex.quote(sys.executable), shlex.quote(str(gate))))
+    stop.chmod(0o755)
+    record: dict = {}
+
+    def writer():
+        try:
+            end = time.monotonic() + 60
+            while not entered.exists() and time.monotonic() < end:
+                time.sleep(0.02)
+            fence = shared / ".reset-in-progress"
+            record["fence_before"] = fence.read_text(encoding="utf-8")
+            env_b = dict(os.environ, WORKSPACE_ROOT=str(other), AGENT_SESSIONS_YAML=str(other / ".mam/agent-sessions.yaml"),
+                         MAM_JOBS_DIR=str(shared), DELEGATE_JOB_LOGS_DIR=str(other / ".mam/delegate_job_logs"),
+                         MAM_ENV_FILE=str(other / ".mam.env.absent"), PYTHONDONTWRITEBYTECODE="1")
+            b = subprocess.run(["/bin/bash", str(other / ".agents/skills/multi-agent-mux-reset/scripts/reset.sh"), "--yes"],
+                               cwd=str(other), env=env_b, stdin=subprocess.DEVNULL, capture_output=True, text=True,
+                               timeout=60)
+            record.update(b_rc=b.returncode, b_out=b.stdout, b_err=b.stderr,
+                          fence_after=fence.read_text(encoding="utf-8") if fence.exists() else None,
+                          b_markers=[p.name for p in (other / ".mam").iterdir() if "progress" in p.name or "guard" in p.name])
+            cli = mam_sandbox / ".agents/skills/multi-agent-mux-delegate-job/scripts/registry.py"
+            res = subprocess.run(
+                [sys.executable, str(cli), "--registry-dir", str(shared), "register",
+                 "--prompt", "external concurrent work", "--agent", "claude-code", "--agent-session", "herdr:" + name],
+                capture_output=True, text=True, timeout=30,
+            )
+            record.update(rc=res.returncode, stderr=res.stderr, jobs=sorted(p.name for p in shared.glob("*.json")))
+        finally:
+            release.touch()
+
+    worker = threading.Thread(target=writer)
+    worker.start()
+    try:
+        res = _run_reset(mam_sandbox, "--yes", env_extra={"MAM_JOBS_DIR": str(shared)})
+    finally:
+        worker.join(timeout=150)
+        _wait_reconcile_idle(mam_sandbox)
+    assert not worker.is_alive()
+    assert record["b_rc"] == 1, record
+    assert "fenced by a reset of %s" % mam_sandbox in record["b_err"] and "nothing was changed" in record["b_err"], record
+    assert record["fence_after"] == record["fence_before"], record
+    assert record["b_markers"] == [], record
+    assert record["rc"] == 1 and "a reset of %s is in progress" % mam_sandbox in record["stderr"], record
+    assert record["jobs"] == [], record
+    if other_has_target:
+        assert other_tf.is_file()
+        assert yaml.safe_load((other / ".mam/agent-sessions.yaml").read_text())["herdr_sessions"][0]["status"] == "running"
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert _result_of(res.stdout, name) == "done"
+    assert not (shared / ".reset-in-progress").exists()