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 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-30 05:23:51 +09:00
1 parent 548c6854ea
commit 568385c77a
6 files changed
+171 -60

No files matched your search

+21 -17
View File
@@ -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 # 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 # may be registered for them while it runs. The reset writes RESET_MARKER into
# publishes .mam/reset-in-progress (pid/lstart, loop_lock.sh's format) and then # every registry its open-job check reads (its own .mam/jobs and MAM_JOBS_DIR,
# takes this registry's lock once before it lists the open jobs; the marker is # and a row's other control_plane.mam_jobs_dir), each under that registry's
# read under the same lock here, so a registration either is written before # lock and before the check: a registration reads it under the same lock, so it
# that listing or sees the marker. The reset's own create (the onboarding job # is either written before the check or refused - by whatever path it names
# of an agent it recreates) passes the reset's pid in MAM_RESET_OWNER_PID. # 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" RESET_OWNER_ENV = "MAM_RESET_OWNER_PID"
@@ -90,15 +94,13 @@ def _lstart(pid: str) -> str:
return " ".join(out.split()) return " ".join(out.split())
def reset_refusal(registry_dir: str) -> Optional[str]: def reset_refusal(registry_dir: str, agent_session: str) -> Optional[str]:
"""Why a registration in ``registry_dir`` is refused, or None: a live """Why registering a job for ``agent_session`` in ``registry_dir`` is
reset of the workspace that owns it (``<ws>/.mam/jobs``) that is not the refused, or None: a live reset fenced this registry (for that session) and
caller's own. Holder states as loop_lock.sh decides them: a dead or the caller is not its own create. Holder states as loop_lock.sh decides
reused pid is stale; ps failing on this process is unknown (refused).""" them: a dead or reused pid is stale; ps failing on this process is unknown
jobs = os.path.abspath(registry_dir) (refused)."""
if os.path.basename(jobs) != "jobs" or os.path.basename(os.path.dirname(jobs)) != ".mam": marker = os.path.join(registry_dir, RESET_MARKER)
return None
marker = os.path.join(os.path.dirname(jobs), "reset-in-progress")
try: try:
with open(marker, encoding="utf-8") as fh: with open(marker, encoding="utf-8") as fh:
fields = dict(line.split("=", 1) for line in fh.read().splitlines() if "=" in line) 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 return None
except (OSError, ValueError) as exc: except (OSError, ValueError) as exc:
return "the reset marker %s cannot be read (%s)" % (marker, 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", "") pid = fields.get("pid", "")
if not pid.isdigit() or os.environ.get(RESET_OWNER_ENV) == pid: if not pid.isdigit() or os.environ.get(RESET_OWNER_ENV) == pid:
return None 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"]): if not cur or (fields.get("lstart") and cur != fields["lstart"]):
return None return None
return ("a reset of %s is in progress (pid %s); its agents are being purged and recreated;" 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( def register_job(
@@ -213,7 +217,7 @@ def register_job(
if dry_run: if dry_run:
record["dry_run"] = True record["dry_run"] = True
with registry_lock(registry_dir): with registry_lock(registry_dir):
refusal = reset_refusal(registry_dir) refusal = reset_refusal(registry_dir, agent_session)
if refusal: if refusal:
raise ResetInProgressError(refusal) raise ResetInProgressError(refusal)
if mqtt_common._job_path(job_id, registry_dir).exists(): if mqtt_common._job_path(job_id, registry_dir).exists():
+47 -7
View File
@@ -18,11 +18,12 @@ import sqlite3
import subprocess import subprocess
import sys import sys
import tempfile 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.base import AGENT_SESSION_ENV, POST_DISCOVERED
from lib_py.agents.registry import agent_of_row, get_adapter, own_key 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.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.update import open_jobs, row_server
from lib_py.verify_session import verify_session_uuid 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 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: def target_id_owned(state: dict, session_name: str) -> bool:
"""The row's recorded post-discovered id still verifies (revalidate) as a """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 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("--workspace", required=True)
s.add_argument("--session", 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 = sub.add_parser("id-owned")
s.add_argument("--workspace", required=True) s.add_argument("--workspace", required=True)
s.add_argument("--session", 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 return 1 if sel["errors"] else 0
if a.cmd == "precheck": 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) lines = precheck(ws, state, os.environ)
for line in lines: for line in lines:
print(line) print(line)
@@ -852,6 +883,15 @@ def main(argv: Optional[List[str]] = None) -> int:
print(line) print(line)
return 1 if lines else 0 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": if a.cmd == "id-owned":
return 0 if target_id_owned(state, a.session) else 1 return 0 if target_id_owned(state, a.session) else 1
@@ -141,8 +141,8 @@ $PY scripts/registry.py status --job <id> --set completed # set status
$PY scripts/registry.py pick --agent-session herdr:claude # claim → running $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 Exit codes: `0` ok, `1` not found / bad status / (`register`) a reset fenced
workspace owning `<ws>/.mam/jobs` is in progress (Issue #78), `3` (`pick`) no this registry (its `.reset-in-progress` names a live reset, Issue #78), `3` (`pick`) no
pending job for that session. pending job for that session.
--- ---
@@ -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. | | 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. | | 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. | | 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`). | | 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). | | 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. | | 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. - **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. - **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. - **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 <WS> 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 <WS> 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. - **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.
@@ -92,9 +92,8 @@ _refuse_locked() {
# The update admission lock is taken exclusively, as update.sh does, while the # The update admission lock is taken exclusively, as update.sh does, while the
# reset publishes its markers: a delegate submit admitted earlier has written # reset publishes its markers: a delegate submit admitted earlier has written
# its job record (the precheck below sees it), and one admitted later sees # 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 # .mam/reset-in-progress and refuses. Any other registration (registry.py
# (registry.py register, the delegation API) reads the marker under the # register, the delegation API) is fenced below.
# registry lock, which the precheck takes before it lists the open jobs.
mkdir -p "$MAM" mkdir -p "$MAM"
exec 7>>"$MAM/update-admission.lock" exec 7>>"$MAM/update-admission.lock"
case "$("$PY" -m lib_py.pane_lock acquire --fd 7 --wait-env MAM_UPDATE_ADMISSION_WAIT_S)" in 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" ;; 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?)" ;; *) _refuse_locked "cannot determine whether a review loop or another reset is running (ps unavailable?)" ;;
esac esac
FENCED=()
_release_reset_lock() { _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 "$RESET_MARKER" || true
mam_release_loop_lock "$MARKER" || true mam_release_loop_lock "$MARKER" || true
} }
@@ -140,6 +144,21 @@ if ! STATE_JSON="$(load_state_json strict)"; then
exit 1 exit 1
fi 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_out=""
pre_rc=0 pre_rc=0
pre_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset precheck --workspace "$WS")" || pre_rc=$? pre_out="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset precheck --workspace "$WS")" || pre_rc=$?
+77 -29
View File
@@ -1655,7 +1655,7 @@ if args[:2] == ["pane", "close"] or args[:1] == ["kill-session"]:
"--agent-session", "herdr:{name_b}", "--prompt", "hi"], "--agent-session", "herdr:{name_b}", "--prompt", "hi"],
cwd={repr(str(mam_sandbox))}, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=120, 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: with open(probe, "w", encoding="utf-8") as fh:
json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr, json.dump({{"rc": res.returncode, "stdout": res.stdout, "stderr": res.stderr,
"marker": os.path.isfile(marker), "jobs": jobs}}, fh) "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 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): @pytest.mark.parametrize("registry", ["canonical", "alias", "custom"])
"""Review round 2 R1 (codex): the documented registry CLI, run by a caller 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 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; before its stop, is refused and registers nothing - whether it names the
the reset's own create still registers its onboarding job.""" 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 import threading
name = "reviewer-claude-01" name = "reviewer-claude-01"
_claude_transcript(mam_sandbox, mam_sandbox, UA) _claude_transcript(mam_sandbox, mam_sandbox, UA)
_mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude") _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude")
_write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)]) _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" stop = mam_sandbox / ".agents/skills/multi-agent-mux-stop/scripts/stop_session.sh"
original = stop.with_name("stop-original.sh") original = stop.with_name("stop-original.sh")
stop.rename(original) 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.write_text("#!/bin/bash\nexec %s %s \"$@\"\n" % (shlex.quote(sys.executable), shlex.quote(str(gate))))
stop.chmod(0o755) stop.chmod(0o755)
jobs_dir = mam_sandbox / ".mam" / "jobs"
record: dict = {} record: dict = {}
def writer(): def writer():
@@ -1831,9 +1845,9 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
time.sleep(0.02) time.sleep(0.02)
record["marker"] = (mam_sandbox / ".mam" / "reset-in-progress").is_file() 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 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( 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], "--prompt", "external concurrent work", "--agent", "claude-code", "--agent-session", "herdr:" + name],
capture_output=True, text=True, timeout=30, 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 = threading.Thread(target=writer)
worker.start() worker.start()
try: try:
res = _run_reset(mam_sandbox, "--yes") res = _run_reset(mam_sandbox, "--yes", env_extra=env_extra)
finally: finally:
worker.join(timeout=90) worker.join(timeout=90)
_wait_reconcile_idle(mam_sandbox) _wait_reconcile_idle(mam_sandbox)
@@ -1858,53 +1872,77 @@ os.execv('/bin/bash', ['bash', {str(original)!r}, *sys.argv[1:]])
assert _result_of(res.stdout, name) == "done" assert _result_of(res.stdout, name) == "done"
onboarding = [json.loads(p.read_text(encoding="utf-8")) for p in jobs_dir.glob("*.json")] 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 [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): def test_register_job_refuses_only_a_live_reset_not_its_own(mam_sandbox: Path, monkeypatch):
from lib_py.delegation import registry from lib_py.delegation import registry
jobs_dir = str(mam_sandbox / ".mam" / "jobs") jobs_dir = mam_sandbox / ".mam" / "jobs"
marker = mam_sandbox / ".mam" / "reset-in-progress" 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) monkeypatch.delenv(registry.RESET_OWNER_ENV, raising=False)
holder = subprocess.Popen(["sleep", "120"]) holder = subprocess.Popen(["sleep", "120"])
try: try:
_live_marker(marker, holder) _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): 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) registry.register_job(prompt="p", agent_session="herdr:b", registry_dir=str(d))
assert not list(Path(jobs_dir).glob("*.json")) # Another control plane's registry refuses only the reset's sessions
# Another control plane's registry is not this workspace's .mam/jobs assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=str(other))
assert registry.reset_refusal(str(mam_sandbox / "elsewhere")) is None assert not list(jobs_dir.glob("*.json"))
monkeypatch.setenv(registry.RESET_OWNER_ENV, str(holder.pid)) 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) monkeypatch.delenv(registry.RESET_OWNER_ENV)
finally: finally:
holder.kill() holder.kill()
holder.wait() holder.wait()
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))
assert len(list(Path(jobs_dir).glob("*.json"))) == 2 assert len(list(jobs_dir.glob("*.json"))) == 2
def test_precheck_waits_for_a_registration_holding_the_registry_lock(tmp_path: Path): def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path: Path):
"""R1 lock order: a registration that read no reset marker holds the registry """R1 lock order and scope: a registration that read no marker holds the
lock until its record is written; the precheck, run after the marker is registry lock until its record is written; fence writes the marker under
published, takes that lock before it lists the open jobs, so it sees the job.""" 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 import fcntl
ws = tmp_path / "ws" ws = tmp_path / "ws"
jobs_dir = ws / ".mam" / "jobs" jobs_dir = ws / ".mam" / "jobs"
jobs_dir.mkdir(parents=True) jobs_dir.mkdir(parents=True)
state = {"herdr_sessions": [_row("reviewer-claude-01", "claude", ws, own_id=UA)]} (ws / "alias").symlink_to(jobs_dir)
env = dict(os.environ, PYTHONPATH=str(SKILLS), PYTHONDONTWRITEBYTECODE="1", MAM_STATE_JSON=json.dumps(state)) other = tmp_path / "cp" / "jobs"
env.pop("MAM_JOBS_DIR", None) 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:
_live_marker(ws / ".mam" / "reset-in-progress", holder)
lock = open(jobs_dir / ".lock", "a") lock = open(jobs_dir / ".lock", "a")
fcntl.flock(lock, fcntl.LOCK_EX) fcntl.flock(lock, fcntl.LOCK_EX)
try: try:
proc = subprocess.Popen( proc = subprocess.Popen(
[sys.executable, "-m", "lib_py.reset", "precheck", "--workspace", str(ws)], [sys.executable, "-m", "lib_py.reset", "fence", "--workspace", str(ws)],
env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
) )
time.sleep(1.5) time.sleep(1.5)
assert proc.poll() is None, proc.communicate() assert proc.poll() is None, proc.communicate()
assert not (jobs_dir / ".reset-in-progress").exists()
(jobs_dir / "job78locked.json").write_text( (jobs_dir / "job78locked.json").write_text(
json.dumps({"job_id": "job78locked", "agent_session": "herdr:reviewer-claude-01", "status": "pending"}), json.dumps({"job_id": "job78locked", "agent_session": "herdr:reviewer-claude-01", "status": "pending"}),
encoding="utf-8", encoding="utf-8",
@@ -1913,8 +1951,18 @@ def test_precheck_waits_for_a_registration_holding_the_registry_lock(tmp_path: P
fcntl.flock(lock, fcntl.LOCK_UN) fcntl.flock(lock, fcntl.LOCK_UN)
lock.close() lock.close()
out, err = proc.communicate(timeout=30) out, err = proc.communicate(timeout=30)
assert proc.returncode == 1, out + err assert proc.returncode == 0, out + err
assert "job job78locked" in out 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:
holder.kill()
holder.wait()
@pytest.mark.parametrize("artifact_cwd", ["own", "foreign"]) @pytest.mark.parametrize("artifact_cwd", ["own", "foreign"])