fix(reset,stop): warn before a reset purges, skip and prune stale registries, refuse a guessed purge

reset.sh without --yes says the conversations are deleted for good before it
asks; without a terminal the refusal says so too. A registry that cannot be
created is skipped, since nothing can be registered there, and the directories
the fence created are removed on exit when still empty. stop --purge-conversation
refuses a conversation the row does not record instead of deleting the folder's
newest one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff

tests/data/skill_code_blocks.json

@@ -629,7 +629,7 @@
     "other"
    ],
    "passthrough": {
-    "1": "multi-agent-mux-stop/scripts/stop_session.sh:389"
+    "1": "multi-agent-mux-stop/scripts/stop_session.sh:394"
    }
   },
   "multi-agent-mux-create/scripts/create_session.sh": {

tests/test_issue63_same_cwd_ids.py

@@ -622,6 +622,35 @@ def test_rev2_claude_unpinned_sibling_row_skips_creator_claimed_id(mam_sandbox,
 
 # --------------------------------------------------------------------------- Done Criterion 6 (id_order & create/resume wait)
 
+@pytest.mark.parametrize("recorded", [None, UB])
+def test_purge_refuses_a_conversation_the_row_does_not_record(mam_sandbox, mock_herdr, mock_agents, recorded):
+    """A pre-assigned row with no usable id: the folder's newest conversation may be
+    anyone's, so a purge refuses and changes nothing. A plain stop records it, and
+    the purge after that (a deliberate second step) deletes it."""
+    ws = mam_sandbox / "claude-guess-ws"
+    ws.mkdir(parents=True, exist_ok=True)
+    epoch = int(time.time()) - 100
+    transcript = _claude_transcript(mam_sandbox / ".claude" / "projects", ws, UA)
+    name = "claude-guess-ws-creator-claude"
+    _mock_herdr_agent(mock_herdr, name, ws, 5001, epoch, agent="claude")
+    _write_rows(mam_sandbox, [_row(name, "claude", ws, epoch, pid=5001, claude_session_id_own=recorded)])
+
+    res = _run_stop(mam_sandbox, name, "claude", purge=True)
+    assert res.returncode == 1, res.stdout + res.stderr
+    assert "refusing to purge" in res.stderr and UA in res.stderr
+    assert transcript.exists()
+    row = _read_rows(mam_sandbox)[name]
+    assert row["status"] == "running" and row["claude_session_id_own"] == recorded
+    assert name in json.loads(Path(mock_herdr).read_text())["agents"]
+
+    if recorded is None:
+        assert _run_stop(mam_sandbox, name, "claude").returncode == 0
+        assert _read_rows(mam_sandbox)[name]["claude_session_id_own"] == UA
+        purge = _run_stop(mam_sandbox, name, "claude", purge=True)
+        assert purge.returncode == 0, purge.stdout + purge.stderr
+        assert not transcript.exists()
+
+
 def test_c6_create_waits_or_refuses_while_same_cwd_post_discovered_row_has_no_id(
     mam_sandbox, mock_herdr, mock_agents
 ):

tests/test_issue78_reset.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import ast
 import json
 import os
+import pty
 import re
 import shlex
 import subprocess
@@ -330,6 +331,7 @@ def _run_reset(
     *args: str,
     stdin_text: Optional[str] = None,
     env_extra: Optional[dict] = None,
+    tty_answer: Optional[str] = None,
 ) -> subprocess.CompletedProcess:
     script = sandbox / ".agents" / "skills" / "multi-agent-mux-reset" / "scripts" / "reset.sh"
     env = dict(
@@ -344,6 +346,22 @@ def _run_reset(
     )
     if env_extra:
         env.update(env_extra)
+    if tty_answer is not None:
+        master, slave = pty.openpty()
+        try:
+            os.write(master, (tty_answer + "\n").encode())
+            return subprocess.run(
+                ["bash", str(script), *args],
+                cwd=str(sandbox),
+                env=env,
+                stdin=slave,
+                capture_output=True,
+                text=True,
+                timeout=180,
+            )
+        finally:
+            os.close(master)
+            os.close(slave)
     return subprocess.run(
         ["bash", str(script), *args],
         cwd=str(sandbox),
@@ -778,6 +796,31 @@ def test_no_yes_non_tty_exits_3_without_touching_locks(mam_sandbox: Path, mock_h
     assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
 
 
+@pytest.mark.parametrize("answer,purged", [("n", False), ("", False), ("y", True)])
+def test_no_yes_on_a_terminal_warns_of_the_deletion_and_asks(
+    mam_sandbox: Path, mock_herdr: Path, mock_agents, answer: str, purged: bool
+):
+    """Without --yes on a terminal the reset says the conversations are deleted for
+    good and asks; only an explicit yes purges.
+    """
+    name = "reviewer-claude-01"
+    transcript = _claude_transcript(mam_sandbox, mam_sandbox, UA)
+    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude", pane_id="w1:p_a")
+    _write_rows(mam_sandbox, [_row(name, "claude", mam_sandbox, own_id=UA)])
+
+    res = _run_reset(mam_sandbox, tty_answer=answer)
+    _wait_reconcile_idle(mam_sandbox)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert "permanently deletes the conversations" in res.stdout
+    assert "they cannot be recovered" in res.stdout
+    assert "Really purge and recreate 1 agents? [y/N]" in res.stderr
+    assert transcript.exists() is not purged
+    assert ("Reset cancelled; nothing was changed" in res.stdout) is not purged
+    assert not (mam_sandbox / ".mam" / "reset-spec.json").exists()
+    assert not (mam_sandbox / ".mam" / "loop-guard-active").exists()
+
+
 def test_dry_run_changes_nothing_in_mam_tree(mam_sandbox: Path, mock_herdr: Path, mock_agents):
     """§5.2-1 & §5.2-4: --dry-run prints targets (with all spec fields), skipped rows,
     and orchestrator_uuids, exits 0, and leaves the entire .mam tree (names, sizes,
@@ -1964,7 +2007,13 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
             lock.close()
         out, err = proc.communicate(timeout=30)
         assert proc.returncode == 0, out + err
-        assert out.splitlines() == ["fenced " + os.path.realpath(p) for p in (jobs_dir, other, absent)]
+        assert out.splitlines() == [
+            "fenced " + os.path.realpath(jobs_dir),
+            "fenced " + os.path.realpath(other),
+            "created " + str(absent),
+            "created " + str(absent.parent),
+            "fenced " + os.path.realpath(absent),
+        ]
         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
@@ -1980,6 +2029,55 @@ def test_fence_waits_for_a_registration_and_covers_every_registry_read(tmp_path:
         holder.wait()
 
 
+def test_stale_row_registries_neither_block_the_reset_nor_stay_behind(
+    mam_sandbox: Path, mock_herdr: Path, mock_agents
+):
+    """A row's registry that cannot be created is skipped; one the fence created is
+    removed at the end, unless something landed in it; an existing one stays."""
+    name = "reviewer-claude-01"
+    transcript = _claude_transcript(mam_sandbox, mam_sandbox, UA)
+    _mock_herdr_agent(mock_herdr, name, mam_sandbox, agent="claude", pane_id="w1:p_a")
+    blocker = mam_sandbox / "a-file"
+    blocker.write_text("x", encoding="utf-8")
+    gone = mam_sandbox / "gone-ws" / ".mam" / "jobs"
+    existing = mam_sandbox / "kept-ws" / "jobs"
+    existing.mkdir(parents=True)
+    rows = [_row(name, "claude", mam_sandbox, own_id=UA)]
+    for i, d in enumerate((blocker / "jobs", gone, existing)):
+        rows.append(_row("old-claude-%d" % i, "claude", mam_sandbox, status="stopped", own_id=None))
+        rows[-1]["control_plane"] = {"mam_jobs_dir": str(d)}
+    _write_rows(mam_sandbox, rows)
+
+    res = _run_reset(mam_sandbox, "--yes")
+    _wait_reconcile_idle(mam_sandbox)
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert not transcript.exists()
+    assert not (mam_sandbox / "gone-ws").exists()
+    assert existing.is_dir() and not (existing / ".reset-in-progress").exists()
+    assert blocker.read_text(encoding="utf-8") == "x"
+
+
+def test_prune_created_keeps_a_directory_with_anything_in_it(tmp_path: Path):
+    used = tmp_path / "used" / "jobs"
+    empty = tmp_path / "empty" / "jobs"
+    for d in (used, empty):
+        d.mkdir(parents=True)
+        (d / ".lock").write_text("", encoding="utf-8")
+    (used / "job1.json").write_text("{}", encoding="utf-8")
+    env = dict(os.environ, PYTHONPATH=str(SKILLS), PYTHONDONTWRITEBYTECODE="1", MAM_STATE_JSON="{}")
+
+    res = subprocess.run(
+        [sys.executable, "-m", "lib_py.reset", "prune-created", "--workspace", str(tmp_path),
+         str(used), str(used.parent), str(empty), str(empty.parent), str(tmp_path / "already-gone")],
+        env=env, capture_output=True, text=True, timeout=30,
+    )
+
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert (used / "job1.json").exists() and (used / ".lock").exists()
+    assert not empty.parent.exists()
+
+
 @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