diff --git a/.agents/skills/lib_py/reset.py b/.agents/skills/lib_py/reset.py index d36f86d..6aaf6e9 100644 --- a/.agents/skills/lib_py/reset.py +++ b/.agents/skills/lib_py/reset.py @@ -21,7 +21,7 @@ import tempfile from typing import Dict, List, Mapping, Optional, Tuple from lib_py.agents.base import AGENT_SESSION_ENV -from lib_py.agents.registry import agent_of_row, own_key +from lib_py.agents.registry import agent_of_row, get_adapter, own_key from lib_py.update import open_jobs, row_server SPEC_VERSION = 1 @@ -227,15 +227,57 @@ def read_spec(workspace: str) -> dict: role = item.get("role") if not (isinstance(name, str) and name and isinstance(agent, str) and agent and isinstance(role, str) and role): raise SpecError("spec file %s has an incomplete agent entry; check it, then remove it by hand" % path) - try: - own_key(agent) - except KeyError: + adapter = get_adapter(agent) + if adapter is None or adapter.name != agent or not own_key(agent): raise SpecError( "spec file %s has unknown agent %r; check it, then remove it by hand" % (path, agent) ) return doc +def record_old_id(workspace: str, state: dict, session_name: str) -> Optional[str]: + """Persist a newly captured conversation id onto an entry with old_id=None. + + Called after step 1 (normal stop) captures a POST_DISCOVERED session's id + onto the original instance's registry row, so a subsequent resume + rerun + cannot mistake the original conversation for a new one (review F1). + Reads the id from `state` (never from command-line args). + """ + ws_real = _real(workspace) + spec = read_spec(ws_real) + rows_by_name = { + str(s.get("name")): s + for s in (state.get("herdr_sessions") or []) + if isinstance(s, dict) and s.get("name") + } + row = rows_by_name.get(session_name) + if row is None: + return None + changed = False + recorded: Optional[str] = None + for entry in spec["agents"]: + if str(entry.get("name") or "") != session_name: + continue + raw_old = entry.get("old_id") + old_id = str(raw_old).strip() if isinstance(raw_old, str) and raw_old.strip() else None + if old_id: + return old_id + cur_created_at = str(row.get("herdr_session_created_at") or "") + old_created_at = str(entry.get("old_created_at") or "") + if cur_created_at != old_created_at: + return None + key = own_key(entry["agent"]) + raw_cur = row.get(key) + cur_id = str(raw_cur).strip() if isinstance(raw_cur, str) and raw_cur.strip() else None + if cur_id: + entry["old_id"] = cur_id + recorded = cur_id + changed = True + if changed: + write_spec(ws_real, spec["agents"]) + return recorded + + def new_running_rows(workspace: str, state: dict, spec_agents: List[dict]) -> List[str]: """Names of running control-root rows not present in spec_agents.""" known = {str(a.get("name") or "") for a in spec_agents} @@ -652,6 +694,10 @@ def main(argv: Optional[List[str]] = None) -> int: s.add_argument("--session", default="") s.add_argument("--warn-new", action="store_true") + s = sub.add_parser("record-id") + s.add_argument("--workspace", required=True) + s.add_argument("--session", required=True) + s = sub.add_parser("table") s.add_argument("--workspace", required=True) s.add_argument("--skipped", action="append", default=[]) @@ -731,6 +777,14 @@ def main(argv: Optional[List[str]] = None) -> int: print(format_entry_line(e, rows_by_name.get(e["name"]))) return 0 + if a.cmd == "record-id": + try: + record_old_id(ws, state, a.session) + except (SpecError, OSError) as exc: + print("ERROR: %s" % exc, file=sys.stderr) + return 1 + return 0 + if a.cmd == "table": try: spec = read_spec(ws) diff --git a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh index 833dbb7..5963263 100755 --- a/.agents/skills/multi-agent-mux-reset/scripts/reset.sh +++ b/.agents/skills/multi-agent-mux-reset/scripts/reset.sh @@ -104,7 +104,15 @@ esac _release_reset_lock() { mam_release_loop_lock "$MARKER" || true } -trap _release_reset_lock EXIT INT TERM HUP +_on_reset_signal() { + local sig_rc="$1" + _release_reset_lock + exit "$sig_rc" +} +trap _release_reset_lock EXIT +trap '_on_reset_signal 130' INT +trap '_on_reset_signal 143' TERM +trap '_on_reset_signal 129' HUP mam_update_admit_release STATE_JSON="" @@ -199,6 +207,12 @@ for target_name in ${ENTRY_NAMES[@]+"${ENTRY_NAMES[@]}"}; do FAILED_SPEC="$R_NAME:stop:1" break fi + if [ -z "$R_OLD_ID" ]; then + MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset record-id --workspace "$WS" --session "$R_NAME" || { + FAILED_SPEC="$R_NAME:stop:1" + break + } + fi entry_line="$(MAM_STATE_JSON="$STATE_JSON" "$PY" -m lib_py.reset entries --workspace "$WS" --session "$R_NAME")" || { FAILED_SPEC="$R_NAME:stop:1" break diff --git a/tests/test_issue78_hints_codex.py b/tests/test_issue78_hints_codex.py index 50d7546..dfae9f2 100644 --- a/tests/test_issue78_hints_codex.py +++ b/tests/test_issue78_hints_codex.py @@ -336,4 +336,3 @@ def test_codex_purge_removes_tui_thread_capability(tmp_path, monkeypatch, delete assert not rollout.exists() assert adapter.residual_artifacts(conv, ctx) == [] assert str(cap_other) in adapter.residual_artifacts(other, ctx) - diff --git a/tests/test_issue78_reset.py b/tests/test_issue78_reset.py index bba86e1..8befe59 100644 --- a/tests/test_issue78_reset.py +++ b/tests/test_issue78_reset.py @@ -501,6 +501,31 @@ def test_spec_file_write_read_and_validation(tmp_path: Path): with pytest.raises(SpecError, match="check it, then remove it by hand"): read_spec(str(ws)) + # Unknown or non-canonical agent in saved spec (review P2) + for bad_agent in ("not-an-agent", "CLAUDE"): + spath.write_text( + json.dumps( + { + "version": 1, + "workspace": str(ws), + "agents": [ + { + "name": "reviewer-bad-01", + "agent": bad_agent, + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + } + ], + } + ), + encoding="utf-8", + ) + with pytest.raises(SpecError, match="unknown agent"): + read_spec(str(ws)) + # Corrupt JSON spath.write_text("{corrupt", encoding="utf-8") with pytest.raises(SpecError, match="check it, then remove it by hand"): @@ -646,6 +671,45 @@ def test_entries_5_states_and_resumed_row_9b(tmp_path: Path): assert rows_by_name["s5-done"][1:6] == ["codex", "-", UD, "yes", "yes"] assert rows_by_name["s5-done"][6] == "done" + # Review F1: (a) old_id=None + same instance + id recorded on row -> record-id persists old_id=UA + from lib_py.reset import read_spec + + res_rec = _run_reset_py( + ["record-id", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"], + state=state, + ) + assert res_rec.returncode == 0, res_rec.stderr + spec_after = read_spec(str(ws)) + s2_entry = next(e for e in spec_after["agents"] if e["name"] == "s2-same-inst-newly-recorded") + assert s2_entry["old_id"] == UA + + # Review F1: (b) after resuming that skipped row (new created_at, id still UA), it is classified + # as 'pending' (NOT falsely 'done') and table exits non-zero (4) with NEW!=OLD == 'no' + state["herdr_sessions"][0] = _row( + "s2-same-inst-newly-recorded", + "agy", + ws, + status="running", + created_at="2026-10-02T00:00:00Z", + own_id=UA, + ) + res_f1_entries = _run_reset_py( + ["entries", "--workspace", str(ws), "--session", "s2-same-inst-newly-recorded"], + state=state, + ) + assert res_f1_entries.returncode == 0 + kv_f1 = dict(tok.split("=", 1) for tok in shlex.split(res_f1_entries.stdout.strip())) + assert kv_f1["R_STATE"] == "pending" + assert kv_f1["R_OLD_ID"] == UA + + res_f1_tbl = _run_reset_py(["table", "--workspace", str(ws)], state=state) + assert res_f1_tbl.returncode == 4 + rows_f1 = { + line.split(None, 6)[0]: line.split(None, 6) + for line in res_f1_tbl.stdout.strip().splitlines()[1:] + } + assert rows_f1["s2-same-inst-newly-recorded"][1:7] == ["agy", UA, UA, "yes", "no", "pending"] + # --------------------------------------------------------------------------- Integration tests: reset.sh @@ -802,6 +866,44 @@ def test_precheck_refusals_change_nothing(mam_sandbox: Path, mock_herdr: Path, m assert "you are reviewer-claude-01, one of the targets" in res_self.stderr _assert_nothing_changed() + # (e) Saved spec with valid pending target first and unknown agent second (review P2) + spec_file = mam_sandbox / ".mam" / "reset-spec.json" + bad_spec_text = json.dumps( + { + "version": 1, + "workspace": str(mam_sandbox), + "written_at": "2026-10-29T00:00:00Z", + "agents": [ + { + "name": "reviewer-claude-01", + "agent": "claude", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UA, + "old_created_at": "2026-10-01T00:00:00Z", + }, + { + "name": "reviewer-bad-01", + "agent": "not-an-agent", + "role": "reviewer", + "herdr_server": "default", + "herdr_workspace": "ws", + "old_id": UB, + "old_created_at": "2026-10-01T00:00:00Z", + }, + ], + } + ) + spec_file.write_text(bad_spec_text, encoding="utf-8") + res_bad_spec = _run_reset(mam_sandbox, "--yes") + assert res_bad_spec.returncode == 1 + assert "unknown agent" in res_bad_spec.stderr + assert "check it, then remove it by hand" in res_bad_spec.stderr + assert spec_file.read_text(encoding="utf-8") == bad_spec_text + spec_file.unlink() + _assert_nothing_changed() + def test_full_reset_success_spec_ordering_table_and_loop_block( mam_sandbox: Path, mock_herdr: Path, mock_agents @@ -1059,7 +1161,8 @@ def test_post_discovered_without_recorded_id_8b_and_exit_5( ) spy_log = _install_herdr_spy(mam_sandbox) - # Run 1: step 1 captures UA onto the row; step 3 refuses purge because lsof holds UA + # Run 1: step 1 captures UA onto the row and record-id persists UA into reset-spec.json; + # step 3 refuses purge because lsof holds UA res1 = _run_reset(mam_sandbox, "--yes") _wait_reconcile_idle(mam_sandbox) assert res1.returncode == 4, res1.stdout + res1.stderr @@ -1067,6 +1170,15 @@ def test_post_discovered_without_recorded_id_8b_and_exit_5( assert rows1[name_a]["status"] == "stopped" assert rows1[name_a]["agy_conversation_id_own"] == UA assert a_files[0].exists() + spec1 = json.loads((mam_sandbox / ".mam" / "reset-spec.json").read_text(encoding="utf-8")) + assert spec1["agents"][0]["old_id"] == UA + + # Simulate resuming the skipped session (new created_at, own_id still UA) before Run 2 (review F1): + # because record-id persisted old_id=UA, Run 2 still treats it as 'pending' and purges UA + rows1[name_a]["status"] = "running" + rows1[name_a]["herdr_session_created_at"] = "2026-10-02T00:00:00Z" + _write_rows(mam_sandbox, [rows1[name_a]]) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, pid=holder.pid, epoch=epoch + 10, agent="agy", pane_id="w1:p_a") # Release lsof and Run 2: step 3 purges UA (deletes all UA artifacts!) and creates new agy lsof_exe.unlink() @@ -1090,7 +1202,7 @@ def test_post_discovered_without_recorded_id_8b_and_exit_5( assert _read_spy_log(spy_log) == [] assert f"reset [{name_a}]:" not in res3.stdout - # Record new ID UB onto the recreated row and Run 4 -> exits 0, OLD_ID='-', NEW!=OLD='yes' + # Record new ID UB onto the recreated row and Run 4 -> exits 0, OLD_ID=UA, NEW!=OLD='yes' rows3 = _read_rows(mam_sandbox) rows3[name_a]["agy_conversation_id_own"] = UB _write_rows(mam_sandbox, [rows3[name_a]]) @@ -1100,7 +1212,7 @@ def test_post_discovered_without_recorded_id_8b_and_exit_5( assert res4.returncode == 0, res4.stdout + res4.stderr assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() tbl_line = [l for l in res4.stdout.splitlines() if l.startswith(name_a)][0] - assert tbl_line.split() == [name_a, "agy", "-", UB, "yes", "yes", "done"] + assert tbl_line.split() == [name_a, "agy", UA, UB, "yes", "yes", "done"] def test_post_discovered_uncapturable_id_8b_ii( @@ -1148,3 +1260,157 @@ def test_post_discovered_uncapturable_id_8b_ii( assert not any(c["verb"] in ("pane close", "kill-session") for c in calls_ii) assert [c["args"][2] for c in calls_ii if c["verb"] == "agent start"] == [name_b] assert name_b in _read_rows(mam_sandbox) + + # Recording new ID UB when old_id remained None produces OLD_ID='-' and NEW!=OLD='yes' + rows_ii_after = _read_rows(mam_sandbox) + rows_ii_after[name_b]["agy_conversation_id_own"] = UB + _write_rows(mam_sandbox, [rows_ii_after[name_b]]) + res_ii_done = _run_reset(mam_sandbox, "--yes") + _wait_reconcile_idle(mam_sandbox) + assert res_ii_done.returncode == 0, res_ii_done.stdout + res_ii_done.stderr + assert not (mam_sandbox / ".mam" / "reset-spec.json").exists() + tbl_ii = [l for l in res_ii_done.stdout.splitlines() if l.startswith(name_b)][0] + assert tbl_ii.split() == [name_b, "agy", "-", UB, "yes", "yes", "done"] + + +@pytest.mark.parametrize( + "sig_name,expected_rc", + [("SIGTERM", 143), ("SIGINT", 130), ("SIGHUP", 129)], +) +def test_signal_interrupts_reset_preserves_spec_and_holds_lock_until_child_exits( + mam_sandbox: Path, mock_herdr: Path, mock_agents, sig_name: str, expected_rc: int +): + """Review P1 / N1: + When SIGTERM, SIGINT, or SIGHUP arrives while an in-flight stop/create child is running: + - reset.sh keeps holding .mam/loop-guard-active (so a competing run_loop.sh is still refused) + until the in-flight child finishes; + - once the child finishes, reset.sh immediately exits with the signal's nonzero code (143/130/129), + releasing .mam/loop-guard-active on exit, preserving .mam/reset-spec.json, and NOT continuing + to purge/create the current or subsequent targets. + """ + import signal + + sig = getattr(signal, sig_name) + name_a = "reviewer-claude-01" + name_b = "reviewer-claude-02" + tf_a = _claude_transcript(mam_sandbox, mam_sandbox, UA) + tf_b = _claude_transcript(mam_sandbox, mam_sandbox, UB) + _mock_herdr_agent(mock_herdr, name_a, mam_sandbox, agent="claude", pane_id="w1:p_a") + _mock_herdr_agent(mock_herdr, name_b, mam_sandbox, agent="claude", pane_id="w1:p_b") + _write_rows( + mam_sandbox, + [ + _row(name_a, "claude", mam_sandbox, own_id=UA), + _row(name_b, "claude", mam_sandbox, own_id=UB), + ], + ) + + # Wrap bin/herdr so the first kill-session blocks until release_gate is created + bin_dir = mam_sandbox / "bin" + herdr_bin = bin_dir / "herdr" + herdr_real = bin_dir / "herdr.real" + if not herdr_real.exists(): + herdr_bin.rename(herdr_real) + entered_gate = mam_sandbox / "child_entered" + release_gate = mam_sandbox / "child_release" + spy_log = mam_sandbox / "herdr_spy.jsonl" + + herdr_bin.write_text( + f"""#!{sys.executable} +import json, os, sys, time + +real = {repr(str(herdr_real))} +entered = {repr(str(entered_gate))} +release = {repr(str(release_gate))} +spy_log = {repr(str(spy_log))} + +raw_args = sys.argv[1:] +args = list(raw_args) +while args and args[0] in ("-L", "--server", "-s", "--session"): + args = args[2:] if len(args) > 1 else args[1:] +verb = args[0] if (args and args[0] in ("kill-session", "has-session", "list-panes", "capture-pane", "send-keys", "ls")) else (" ".join(args[:2]) if len(args) >= 2 else "") +with open(spy_log, "a", encoding="utf-8") as lf: + lf.write(json.dumps({{"verb": verb, "args": args}}) + "\\n") + +if verb == "kill-session" and not os.path.exists(entered): + with open(entered, "w", encoding="utf-8") as fh: + fh.write("1") + deadline = time.time() + 15.0 + while not os.path.exists(release) and time.time() < deadline: + time.sleep(0.05) + +os.execv(real, [real, *raw_args]) +""", + encoding="utf-8", + ) + herdr_bin.chmod(0o755) + + script = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh" + env = dict( + os.environ, + SKS_QUIESCENT_TRIES="0", + SKS_DIALOG_TIMEOUT="0", + SKS_BUSY_TIMEOUT="0", + MAM_ONBOARD_ACK_S="3", + MAM_ONBOARD_WAIT_S="3", + MAM_NO_SUBSCRIBER_DAEMON="1", + PYTHONDONTWRITEBYTECODE="1", + ) + proc = subprocess.Popen( + ["bash", str(script), "--yes"], + cwd=str(mam_sandbox), + env=env, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + try: + deadline = time.time() + 15.0 + while not entered_gate.exists() and time.time() < deadline: + assert proc.poll() is None, "reset.sh exited before reaching kill-session" + time.sleep(0.05) + assert entered_gate.exists(), "timed out waiting for kill-session child" + + # Signal the reset parent while the foreground child is still blocked + proc.send_signal(sig) + time.sleep(0.2) + + # While the child is still running, reset.sh has not exited and still holds the loop lock + assert proc.poll() is None + assert (mam_sandbox / ".mam" / "loop-guard-active").is_file() + if sig_name == "SIGTERM": + loop_sh = mam_sandbox / ".agents" / "skills" / "multi-agent-mux-loop" / "scripts" / "run_loop.sh" + competing = subprocess.run( + ["bash", str(loop_sh), "--creator", "x", "--task", "t"], + cwd=str(mam_sandbox), + env=dict(os.environ, MAM_LOOP_NO_FREEZE="1"), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=20, + ) + assert competing.returncode == 1 + assert "Another multi-agent-mux-loop is already running" in (competing.stdout + competing.stderr) + + # Release the child and wait for reset.sh to terminate + release_gate.write_text("1", encoding="utf-8") + stdout, stderr = proc.communicate(timeout=30) + finally: + if proc.poll() is None: + proc.kill() + proc.wait() + _wait_reconcile_idle(mam_sandbox) + + assert proc.returncode == expected_rc, f"rc={proc.returncode}, stdout={stdout}, stderr={stderr}" + assert not (mam_sandbox / ".mam" / "loop-guard-active").exists() + assert (mam_sandbox / ".mam" / "reset-spec.json").is_file() + # Neither transcript was purged, and no create or second-target operation ran + assert tf_a.is_file() + assert tf_b.is_file() + calls = _read_spy_log(spy_log) + assert not any(c["verb"] == "agent start" for c in calls) + assert f"reset [{name_b}]:" not in stdout + assert not any( + l.startswith("reset [") and "--purge-conversation" in l for l in stdout.splitlines() + )