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>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-30 08:21:40 +09:00
1 parent 568385c77a
commit b49314c3d0
6 files changed
+205 -66

No files matched your search

+27 -14
View File
@@ -94,32 +94,45 @@ def _lstart(pid: str) -> str:
return " ".join(out.split())
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)."""
def reset_marker(registry_dir: str) -> Tuple[str, Dict[str, str]]:
"""(holder state, fields) of the registry's RESET_MARKER: "absent", or as
loop_lock.sh decides - "stale" (dead or reused pid), "live", "unknown" (ps
fails on this process; a marker that cannot be read is unknown too)."""
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)
except FileNotFoundError:
return None
return "absent", {}
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
return "unknown", {"error": str(exc)}
pid = fields.get("pid", "")
if not pid.isdigit() or os.environ.get(RESET_OWNER_ENV) == pid:
return None
if not pid.isdigit():
return "stale", fields
if not _lstart(str(os.getpid())):
return "cannot tell whether the reset in %s (pid %s) is live (ps unavailable?)" % (marker, pid)
return "unknown", fields
cur = _lstart(pid)
if not cur or (fields.get("lstart") and cur != fields["lstart"]):
return "stale", fields
return "live", fields
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 (or undecidable) reset fenced this registry, for
that session, and the caller is not its own create."""
state, fields = reset_marker(registry_dir)
if state in ("absent", "stale"):
return None
if "sessions" in fields and agent_session not in fields["sessions"].split(","):
return None
if fields.get("pid") and os.environ.get(RESET_OWNER_ENV) == fields["pid"]:
return None
if state == "unknown":
return "cannot tell whether the reset fencing %s is live (%s)" % (
os.path.join(registry_dir, RESET_MARKER), fields.get("error") or "ps unavailable?")
return ("a reset of %s is in progress (pid %s); its agents are being purged and recreated;"
" register the job when it has finished" % (fields.get("workspace", "?"), pid))
" register the job when it has finished" % (fields.get("workspace", "?"), fields["pid"]))
def register_job(
+26 -22
View File
@@ -23,8 +23,8 @@ 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.delegation.registry import RESET_MARKER, reset_marker
from lib_py.update import jobs_dirs, open_jobs, row_server
from lib_py.verify_session import verify_session_uuid
SPEC_VERSION = 1
@@ -485,35 +485,36 @@ def target_open_jobs(workspace: str, state: dict, env: Mapping[str, str], sessio
return out
class FenceBusy(Exception):
pass
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
"""Write RESET_MARKER into every registry open_jobs reads (jobs_dirs,
created when absent - a registration would create it too), 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."""
and yield its real path as it is written. FenceBusy when one is already
fenced by another live (or undecidable) reset - registries may be shared
between workspaces; its marker is left as it is. This workspace's own
registries refuse every registration; another control plane's 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"))
me = dict(l.split("=", 1) for l in base.splitlines()).get("pid", "")
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:
for d, addressed_only in jobs_dirs(workspace, state, env):
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):
held, fields = reset_marker(d)
if held in ("live", "unknown") and fields.get("pid") != me:
raise FenceBusy(
"the job registry %s is fenced by a reset of %s (pid %s, holder %s); wait for it to end"
% (d, fields.get("workspace", "?"), fields.get("pid", "?"), held))
fd, tmp = tempfile.mkstemp(dir=d, prefix=RESET_MARKER + ".")
with os.fdopen(fd, "w", encoding="utf-8") as fh:
fh.write(text)
fh.write(base + ("sessions=%s\n" % sessions if addressed_only else ""))
os.replace(tmp, os.path.join(d, RESET_MARKER))
yield _real(d)
@@ -886,9 +887,12 @@ def main(argv: Optional[List[str]] = None) -> int:
if a.cmd == "fence":
try:
for d in fence_registries(ws, state, os.environ):
print(d, flush=True)
print("fenced %s" % d, flush=True)
except FenceBusy as exc:
print("refused %s" % exc)
return 1
except OSError as exc:
print("ERROR: cannot fence the job registries (%s)" % exc, file=sys.stderr)
print("refused the job registries cannot be fenced against new registrations (%s)" % exc)
return 1
return 0
+23 -17
View File
@@ -147,27 +147,33 @@ def _is_open(rec: dict) -> bool:
return rec.get("status") not in TERMINAL_STATUSES and rec.get("dry_run") is not True
def open_jobs(workspace: str, state: dict, env: Mapping[str, str]) -> List[dict]:
"""Every open record (not completed/error/cancelled, not dry_run) in this
workspace's .mam/jobs and MAM_JOBS_DIR; in another row's
control_plane.mam_jobs_dir only those addressed to a row of this registry.
A record, or a directory, that cannot be read is open (path only)."""
own, seen = [], set()
for c in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or ""):
if c and _is_jobs_dir(c) and _real(c) not in seen:
def jobs_dirs(workspace: str, state: dict, env: Mapping[str, str]) -> List[Tuple[str, bool]]:
"""(dir, addressed_only) per registry an open-job check reads, once per
real path, whether or not it exists yet: this workspace's .mam/jobs and
MAM_JOBS_DIR, then another row's control_plane.mam_jobs_dir, where only
records addressed to a row of this registry count (addressed_only)."""
rows = state.get("herdr_sessions") or []
cands = [(c, False) for c in (os.path.join(workspace, ".mam", "jobs"), env.get("MAM_JOBS_DIR") or "")]
cands += [((s.get("control_plane") or {}).get("mam_jobs_dir") or "", True) for s in rows]
out, seen = [], set()
for c, addressed_only in cands:
if c and _real(c) not in seen:
seen.add(_real(c))
own.append(c)
out.append((c, addressed_only))
return out
def open_jobs(workspace: str, state: dict, env: Mapping[str, str]) -> List[dict]:
"""Every open record (not completed/error/cancelled, not dry_run) in the
registries jobs_dirs names; in another row's control-plane registry only
those addressed to a row of this registry. A record, or a directory, that
cannot be read is open (path only)."""
rows = state.get("herdr_sessions") or []
names = {"herdr:%s" % s.get("name") for s in rows if s.get("name")}
other = []
for s in rows:
c = (s.get("control_plane") or {}).get("mam_jobs_dir") or ""
if c and _is_jobs_dir(c) and _real(c) not in seen:
seen.add(_real(c))
other.append(c)
out = []
for dirs, addressed_only in ((own, False), (other, True)):
for d in dirs:
for d, addressed_only in jobs_dirs(workspace, state, env):
if not _is_jobs_dir(d):
continue
for path, rec in _job_records(d):
if rec is None:
out.append({"path": path, "unreadable": True, "dir": path == d})
@@ -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. 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. |
| 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`; created when absent), each under that registry's lock, before the check. It refuses, changing nothing, when a registry shared with another workspace is already fenced by that workspace's live reset; it never replaces or removes another reset's marker. 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. |
@@ -150,12 +150,16 @@ fi
fence_out=""
fence_rc=0
fence_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset fence --workspace "$WS")" || fence_rc=$?
fence_refusal=""
while IFS= read -r line; do
[ -n "$line" ] && FENCED+=("$line")
case "$line" in
"fenced "*) FENCED+=("${line#fenced }") ;;
"refused "*) fence_refusal="${line#refused }" ;;
esac
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
echo " - ${fence_refusal:-the job registries could not be fenced against new registrations (lib_py.reset fence rc $fence_rc)}" >&2
exit 1
fi
+122 -10
View File
@@ -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()