fix(reset): refuse every registration during a reset and purge only ids owned by the row's folder (#78)

Review round 2 (codex):
- R1: register_job reads .mam/reset-in-progress under the registry lock and
  refuses while a reset of the workspace owning <ws>/.mam/jobs is live, so the
  public registry.py register CLI and the API no longer bypass the gate. The
  reset's precheck takes that lock once after publishing its marker, so a
  registration is either seen by it or refused. The reset's own create passes
  MAM_RESET_OWNER_PID; the remote runner refuses the dispatch.
- R2: before each purge the reset re-runs stop's revalidate check
  (verify_session_uuid) for a post-discovered id against the row's folder and
  skips the target (id not verified) when it does not verify.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
GodopuandClaude Opus 5.5 committed 2026-10-30 03:45:27 +09:00
1 parent 18ea5f6184
commit 548c6854ea
7 files changed
+287 -8

No files matched your search

@@ -16,6 +16,8 @@ from __future__ import annotations
import argparse import argparse
import json import json
import logging import logging
import os
import subprocess
import sys import sys
import uuid import uuid
from pathlib import Path from pathlib import Path
@@ -65,6 +67,57 @@ def liveness_window(heartbeat_sec: int, heartbeat_misses: int) -> int:
return max(1, int(heartbeat_sec)) * max(1, int(heartbeat_misses)) return max(1, int(heartbeat_sec)) * max(1, int(heartbeat_misses))
# 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.
RESET_OWNER_ENV = "MAM_RESET_OWNER_PID"
class ResetInProgressError(RuntimeError):
pass
def _lstart(pid: str) -> str:
try:
out = subprocess.run(["ps", "-p", pid, "-o", "lstart="], capture_output=True, text=True,
timeout=10).stdout
except (OSError, subprocess.SubprocessError):
return ""
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 (``<ws>/.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")
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
except (OSError, ValueError) as exc:
return "the reset marker %s cannot be read (%s)" % (marker, exc)
pid = fields.get("pid", "")
if not pid.isdigit() or os.environ.get(RESET_OWNER_ENV) == pid:
return None
if not _lstart(str(os.getpid())):
return "cannot tell whether the reset in %s (pid %s) is live (ps unavailable?)" % (marker, pid)
cur = _lstart(pid)
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))
def register_job( def register_job(
prompt: str, prompt: str,
agent: str = "claude-code", agent: str = "claude-code",
@@ -160,6 +213,9 @@ 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)
if refusal:
raise ResetInProgressError(refusal)
if mqtt_common._job_path(job_id, registry_dir).exists(): if mqtt_common._job_path(job_id, registry_dir).exists():
if not supersede_older_attempt: if not supersede_older_attempt:
raise FileExistsError(f"job already exists: {job_id}") raise FileExistsError(f"job already exists: {job_id}")
@@ -556,6 +612,7 @@ def main(argv: Optional[List[str]] = None) -> int:
if args.parent_job and not mqtt_common._job_path(args.parent_job, rd).exists(): if args.parent_job and not mqtt_common._job_path(args.parent_job, rd).exists():
print(f"unknown parent job {args.parent_job}", file=sys.stderr) print(f"unknown parent job {args.parent_job}", file=sys.stderr)
return 1 return 1
try:
job_id = register_job( job_id = register_job(
prompt=_prompt_from_args(parser, args), prompt=_prompt_from_args(parser, args),
agent=args.agent, agent=args.agent,
@@ -581,6 +638,9 @@ def main(argv: Optional[List[str]] = None) -> int:
parent_job=args.parent_job, parent_job=args.parent_job,
dry_run=args.dry_run, dry_run=args.dry_run,
) )
except ResetInProgressError as exc:
print("ERROR: %s" % exc, file=sys.stderr)
return 1
print(job_id) print(job_id)
return 0 return 0
@@ -221,6 +221,9 @@ def mirror_job(d: Dict[str, Any], registry_dir: str, session: str) -> Optional[s
except FileExistsError as exc: # lost the race to a newer (or equal) attempt under the lock except FileExistsError as exc: # lost the race to a newer (or equal) attempt under the lock
logger.warning("refusing dispatch for %s: %s", jid, exc) logger.warning("refusing dispatch for %s: %s", jid, exc)
return None return None
except registry.ResetInProgressError as exc: # Issue #78: this workspace is being reset
logger.warning("refusing dispatch for %s: %s", jid, exc)
return None
brief_dir = os.path.join(registry_dir, jid) brief_dir = os.path.join(registry_dir, jid)
os.makedirs(brief_dir, exist_ok=True) os.makedirs(brief_dir, exist_ok=True)
with open(os.path.join(brief_dir, "brief.md"), "w", encoding="utf-8") as fh: with open(os.path.join(brief_dir, "brief.md"), "w", encoding="utf-8") as fh:
+33
View File
@@ -22,7 +22,9 @@ from typing import Dict, 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.update import open_jobs, row_server from lib_py.update import open_jobs, row_server
from lib_py.verify_session import verify_session_uuid
SPEC_VERSION = 1 SPEC_VERSION = 1
SPEC_REL = os.path.join(".mam", "reset-spec.json") SPEC_REL = os.path.join(".mam", "reset-spec.json")
@@ -482,6 +484,24 @@ def target_open_jobs(workspace: str, state: dict, env: Mapping[str, str], sessio
return out return out
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
an id. stop's own --purge-conversation also removes an id that no longer
verifies (a hand-run orphan cleanup); the reset purges only an owned one.
stop looks a pre-assigned id up in the cwd itself, so those pass."""
row = _rows_by_name(state).get(session_name)
agent = agent_of_row(row) if row else None
if not agent:
return False
if get_adapter(agent).id_lifecycle != POST_DISCOVERED:
return True
raw = row.get(own_key(agent))
own = raw.strip() if isinstance(raw, str) else ""
cwd = str((row.get("pane") or {}).get("cwd") or "")
return bool(own and cwd and verify_session_uuid(cwd, agent, own, row, mode="revalidate"))
# --------------------------------------------------------------------------- D1: entries shell lines # --------------------------------------------------------------------------- D1: entries shell lines
def format_entry_line(entry: dict, row: Optional[dict]) -> str: def format_entry_line(entry: dict, row: Optional[dict]) -> str:
@@ -737,6 +757,10 @@ 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("id-owned")
s.add_argument("--workspace", required=True)
s.add_argument("--session", required=True)
s = sub.add_parser("record-id") s = sub.add_parser("record-id")
s.add_argument("--workspace", required=True) s.add_argument("--workspace", required=True)
s.add_argument("--session", required=True) s.add_argument("--session", required=True)
@@ -790,6 +814,12 @@ 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)
@@ -822,6 +852,9 @@ 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 == "id-owned":
return 0 if target_id_owned(state, a.session) else 1
if a.cmd == "record-id": if a.cmd == "record-id":
try: try:
record_old_id(ws, state, a.session) record_old_id(ws, state, a.session)
@@ -141,8 +141,9 @@ $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, `3` (`pick`) no pending job for Exit codes: `0` ok, `1` not found / bad status / (`register`) a reset of the
that session. workspace owning `<ws>/.mam/jobs` is in progress (Issue #78), `3` (`pick`) no
pending job for that session.
--- ---
@@ -23,10 +23,10 @@ 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; `multi-agent-mux-delegate-job submit` refuses on the reset marker. | | 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`. |
| 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, ③ `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 (`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. |
| Exit | Meaning | | Exit | Meaning |
|---|---| |---|---|
@@ -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` refuses with `a reset of <WS> is in progress` until the reset ends; submit again afterwards. A job registered for a target by other means after the precheck makes that target `job open` (skipped, not cancelled). - **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.
- **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,7 +92,9 @@ _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. # .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.
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
@@ -255,6 +257,11 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do
SKIPPED_ARGS+=(--skipped "$R_NAME:id_unverified") SKIPPED_ARGS+=(--skipped "$R_NAME:id_unverified")
continue continue
fi fi
if ! MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset id-owned --workspace "$WS" --session "$R_NAME"; then
echo "skipped $R_NAME: its recorded id $R_ROW_ID does not verify as a conversation of its folder, so it may be another folder's and nothing is purged; check it, purge it by hand if it is this agent's, and rerun the reset"
SKIPPED_ARGS+=(--skipped "$R_NAME:id_unverified")
continue
fi
echo "reset [$R_NAME]: ${STOP_CMD[*]} --purge-conversation --yes" echo "reset [$R_NAME]: ${STOP_CMD[*]} --purge-conversation --yes"
purge_rc=0 purge_rc=0
@@ -283,7 +290,8 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do
echo "reset [$R_NAME]: ${CREATE_CMD[*]}" echo "reset [$R_NAME]: ${CREATE_CMD[*]}"
create_rc=0 create_rc=0
"${CREATE_CMD[@]}" || create_rc=$? # Its onboarding job is the one registration the reset marker lets through.
MAM_RESET_OWNER_PID="$$" "${CREATE_CMD[@]}" || create_rc=$?
if [ "$create_rc" -ne 0 ]; then if [ "$create_rc" -ne 0 ]; then
FAILED_SPEC="$R_NAME:create:$create_rc" FAILED_SPEC="$R_NAME:create:$create_rc"
break break
+174
View File
@@ -1788,3 +1788,177 @@ def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path):
holder.wait() holder.wait()
res_stale = admit() res_stale = admit()
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):
"""Review round 2 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."""
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)])
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() + 60
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)
jobs_dir = mam_sandbox / ".mam" / "jobs"
record: dict = {}
def writer():
try:
end = time.monotonic() + 60
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"))
registry = 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",
"--prompt", "external concurrent work", "--agent", "claude-code", "--agent-session", "herdr:" + name],
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")))
finally:
release.touch()
worker = threading.Thread(target=writer)
worker.start()
try:
res = _run_reset(mam_sandbox, "--yes")
finally:
worker.join(timeout=90)
_wait_reconcile_idle(mam_sandbox)
assert not worker.is_alive()
assert record["marker"] is True, record
assert record["rc"] == 1, record
assert "a reset of %s is in progress" % mam_sandbox in record["stderr"], record
assert record["stdout"] == "" and record["after"] == record["before"], record
assert res.returncode == 0, res.stdout + res.stderr
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
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"
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
monkeypatch.setenv(registry.RESET_OWNER_ENV, str(holder.pid))
assert registry.register_job(prompt="p", agent_session="herdr:x", registry_dir=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
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."""
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)
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",
)
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
@pytest.mark.parametrize("artifact_cwd", ["own", "foreign"])
def test_post_discovered_id_is_purged_only_when_it_verifies_for_its_folder(
mam_sandbox: Path, mock_herdr: Path, mock_agents, monkeypatch, artifact_cwd: str
):
"""Review round 2 R2 (codex): a recorded codex id whose rollout names another
folder is not purged (stop's own unverified-own purge is for a hand-run
cleanup); the same rollout naming the row's folder is."""
from lib_py.agents.base import DiscoveryContext
from lib_py.agents.registry import get_adapter
foreign = mam_sandbox / "another-workspace"
foreign.mkdir()
cwd = mam_sandbox if artifact_cwd == "own" else foreign
home = mam_sandbox / ".codex"
monkeypatch.setenv("CODEX_HOME", str(home))
rollout = home / "sessions/2026/10/30" / ("rollout-2026-10-30T00-00-00-%s.jsonl" % UB)
rollout.parent.mkdir(parents=True)
rollout.write_text(json.dumps({"type": "session_meta", "payload": {"id": UB, "cwd": str(cwd)}}) + "\n")
name = "reviewer-codex-01"
row = _row(name, "codex", mam_sandbox, own_id=UB)
row["session_id_verified"] = False
_mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="codex")
_write_rows(mam_sandbox, [row])
# The CLI delete fails, so the adapter's own fallback removes the rollout
codex = mam_sandbox / "bin" / "codex"
codex.write_text('#!/bin/bash\nif [ "$1" = delete ]; then exit 1; fi\nexit 0\n')
codex.chmod(0o755)
ctx = DiscoveryContext(workspace=str(mam_sandbox), agent_name="codex", home_dir=str(mam_sandbox),
row=row, mode="revalidate")
assert get_adapter("codex").verify_artifact(UB, ctx) is (artifact_cwd == "own")
res = _run_reset(mam_sandbox, "--yes")
_wait_reconcile_idle(mam_sandbox)
if artifact_cwd == "own":
assert res.returncode == 0, res.stdout + res.stderr
assert _result_of(res.stdout, name) == "done"
assert not rollout.exists()
else:
assert res.returncode == 4, res.stdout + res.stderr
assert _result_of(res.stdout, name) == "skipped: id not verified"
assert "--purge-conversation --yes" not in res.stdout
assert rollout.is_file()
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()