fix(reset): record captured old_id in spec, exit on signals, and reject unknown spec agents (#78)

This commit is contained in:
Godopu committed 2026-10-29 12:52:17 +00:00
1 parent fb1895030f
commit b5040cffab
4 files changed
+342 -9

No files matched your search

+58 -4
View File
@@ -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)
@@ -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
-1
View File
@@ -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)
+269 -3
View File
@@ -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()
)