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>

.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 (``<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(
     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
 

.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:

.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)