diff --git a/.agents/skills/lib_py/delegation/registry.py b/.agents/skills/lib_py/delegation/registry.py index 9dff5d8..1722fb9 100755 --- a/.agents/skills/lib_py/delegation/registry.py +++ b/.agents/skills/lib_py/delegation/registry.py @@ -16,6 +16,8 @@ from __future__ import annotations import argparse import json import logging +import os +import subprocess import sys import uuid 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)) +# 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 (``/.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( prompt: str, agent: str = "claude-code", @@ -160,6 +213,9 @@ def register_job( if dry_run: record["dry_run"] = True 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 not supersede_older_attempt: raise FileExistsError(f"job already exists: {job_id}") @@ -556,31 +612,35 @@ def main(argv: Optional[List[str]] = None) -> int: 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) return 1 - job_id = register_job( - prompt=_prompt_from_args(parser, args), - agent=args.agent, - agent_session=args.agent_session, - role=args.role, - timeout_sec=args.timeout, - idle_timeout_sec=args.idle_timeout, - registry_dir=rd, - expected_artifacts=args.artifacts, - bits=args.bits, - auth_token=args.auth_token, - job_type=args.job_type, - reviewer=args.reviewer, - reviewer_session=args.reviewer_session, - max_iterations=args.max_iterations, - heartbeat_sec=args.heartbeat_sec, - heartbeat_misses=args.heartbeat_misses, - ack_timeout_sec=args.ack_timeout, - transport=args.transport, - dispatch_token=args.dispatch_token, - attempt_id=args.attempt_id, - job_id=args.job_id, - parent_job=args.parent_job, - dry_run=args.dry_run, - ) + try: + job_id = register_job( + prompt=_prompt_from_args(parser, args), + agent=args.agent, + agent_session=args.agent_session, + role=args.role, + timeout_sec=args.timeout, + idle_timeout_sec=args.idle_timeout, + registry_dir=rd, + expected_artifacts=args.artifacts, + bits=args.bits, + auth_token=args.auth_token, + job_type=args.job_type, + reviewer=args.reviewer, + reviewer_session=args.reviewer_session, + max_iterations=args.max_iterations, + heartbeat_sec=args.heartbeat_sec, + heartbeat_misses=args.heartbeat_misses, + ack_timeout_sec=args.ack_timeout, + transport=args.transport, + dispatch_token=args.dispatch_token, + attempt_id=args.attempt_id, + job_id=args.job_id, + parent_job=args.parent_job, + dry_run=args.dry_run, + ) + except ResetInProgressError as exc: + print("ERROR: %s" % exc, file=sys.stderr) + return 1 print(job_id) return 0 diff --git a/.agents/skills/lib_py/delegation/remote.py b/.agents/skills/lib_py/delegation/remote.py index 47dcbdf..584f8f7 100644 --- a/.agents/skills/lib_py/delegation/remote.py +++ b/.agents/skills/lib_py/delegation/remote.py @@ -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 logger.warning("refusing dispatch for %s: %s", jid, exc) 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) os.makedirs(brief_dir, exist_ok=True) with open(os.path.join(brief_dir, "brief.md"), "w", encoding="utf-8") as fh: diff --git a/.agents/skills/lib_py/reset.py b/.agents/skills/lib_py/reset.py index 7beaf54..b2a150e 100644 --- a/.agents/skills/lib_py/reset.py +++ b/.agents/skills/lib_py/reset.py @@ -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.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.verify_session import verify_session_uuid SPEC_VERSION = 1 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 +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 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("--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.add_argument("--workspace", 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 if a.cmd == "precheck": + # A registration that read no reset marker finishes writing under this + # lock before the open jobs are listed (registry.reset_refusal). + jobs_dir = os.path.join(ws, ".mam", "jobs") + if os.path.isdir(jobs_dir): + with registry_lock(jobs_dir): + pass lines = precheck(ws, state, os.environ) for line in lines: print(line) @@ -822,6 +852,9 @@ def main(argv: Optional[List[str]] = None) -> int: print(line) 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": try: record_old_id(ws, state, a.session) diff --git a/.agents/skills/multi-agent-mux-delegate-job/registry.md b/.agents/skills/multi-agent-mux-delegate-job/registry.md index 6fe063f..0d70570 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/registry.md +++ b/.agents/skills/multi-agent-mux-delegate-job/registry.md @@ -141,8 +141,9 @@ $PY scripts/registry.py status --job --set completed # set status $PY scripts/registry.py pick --agent-session herdr:claude # claim → running ``` -Exit codes: `0` ok, `1` not found / bad status, `3` (`pick`) no pending job for -that session. +Exit codes: `0` ok, `1` not found / bad status / (`register`) a reset of the +workspace owning `/.mam/jobs` is in progress (Issue #78), `3` (`pick`) no +pending job for that session. --- diff --git a/.agents/skills/multi-agent-mux-reset/SKILL.md b/.agents/skills/multi-agent-mux-reset/SKILL.md index da8ddd5..cb2fa29 100644 --- a/.agents/skills/multi-agent-mux-reset/SKILL.md +++ b/.agents/skills/multi-agent-mux-reset/SKILL.md @@ -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. | | 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; `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`). | -| 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). | -| 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. | +| 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. | | 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. - **Shared stores preserved**: Grok's shared `memory-v2.db` is left intact during purge; only session rows and per-session files are removed. - **Orchestrator hook during reset**: while `.mam/loop-guard-active` is held by the reset, the AGY orchestrator's direct file-edit hook (`loop_delegation_guard.sh`) blocks direct edits until the reset finishes. -- **Job submissions during a reset**: `multi-agent-mux-delegate-job submit` refuses with `a reset of 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 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. diff --git a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh index 7978b7a..7d3afef 100755 --- a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh +++ b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh @@ -92,7 +92,9 @@ _refuse_locked() { # The update admission lock is taken exclusively, as update.sh does, while the # reset publishes its markers: a delegate submit admitted earlier has written # its job record (the precheck below sees it), and one admitted later sees -# .mam/reset-in-progress and refuses. +# .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" exec 7>>"$MAM/update-admission.lock" 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") continue 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" purge_rc=0 @@ -283,7 +290,8 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do echo "reset [$R_NAME]: ${CREATE_CMD[*]}" 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 FAILED_SPEC="$R_NAME:create:$create_rc" break diff --git a/tests/test_issue78_reset.py b/tests/test_issue78_reset.py index fe6d3fa..27f5f9b 100644 --- a/tests/test_issue78_reset.py +++ b/tests/test_issue78_reset.py @@ -1788,3 +1788,177 @@ def test_mam_reset_admit_refuses_only_a_live_reset(mam_sandbox: Path): holder.wait() res_stale = admit() 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()