.agents/skills/lib_py/stall_watch.py

@@ -232,7 +232,8 @@ class StallWatch(object):
         self.once = set()
 
     def _notify(self, detail):
-        """The only publisher: a `progress` event, nothing else."""
+        """The only publisher: a `progress` event, nothing else. False when the
+        registry or the wire refused it (a refused notice withholds the nudge)."""
         cmd = [sys.executable, "-m", "lib_py.delegation.publish_event", "--registry-dir", self.args.registry_dir,
                "--job", self.args.job, "--event", "progress", "--detail", detail, "--attempts", "1"]
         if self.args.expect_dispatch_token:
@@ -245,6 +246,8 @@ class StallWatch(object):
         res = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=120)
         if res.returncode not in (0, 2):
             log.warning("stall_watch: publishing %r failed (rc %d): %s", detail, res.returncode, res.stderr.strip()[-200:])
+            return False
+        return True
 
     def _notify_once(self, key, detail, message):
         if key in self.once:
@@ -260,6 +263,28 @@ class StallWatch(object):
         except (OSError, ValueError):
             return None
 
+    def _dispatch_open(self):
+        """Whether the job is still open *and* still the dispatch we were started
+        for: a `completed` or a re-dispatch that landed during a blocking herdr
+        wait must not be nudged (review R2). A missing record counts as gone;
+        an unreadable one as a transient."""
+        try:
+            with open(os.path.join(self.args.registry_dir, self.args.job + ".json")) as f:
+                rec = json.load(f)
+        except FileNotFoundError:
+            return False
+        except (OSError, ValueError):
+            return True
+        if rec.get("status") in TERMINAL_STATUSES:
+            return False
+        tok = self.args.expect_dispatch_token
+        if tok and rec.get("dispatch_token") and rec.get("dispatch_token") != tok:
+            return False
+        att = self.args.attempt_id
+        if att is not None and rec.get("attempt_id") is not None and int(rec["attempt_id"]) != int(att):
+            return False
+        return True
+
     def run(self) -> int:
         st, sig, agent = self.st, Signals(), self.args.agent
         if not st.reliable:
@@ -268,7 +293,13 @@ class StallWatch(object):
         fail_since, idle_since = None, None
         while True:
             sig.job_status = self._job_status()
-            if sig.job_status in TERMINAL_STATUSES:
+            if sig.job_status in TERMINAL_STATUSES or not self._dispatch_open():
+                return RC_TERMINAL
+            if self.args.watch_pid and not _alive(self.args.watch_pid):
+                # The wrapper is gone (killed, crashed): nobody will stop us, so
+                # stop ourselves. Checked between herdr calls, so at most one
+                # wait (<= 60 s) late.
+                log.info("stall_watch: wrapper pid %d is gone; stall watch stops", self.args.watch_pid)
                 return RC_TERMINAL
             if sig.agent_status is None:
                 until, tmo = watcher.VALID_STATES, 1000  # learn the current state
@@ -298,6 +329,9 @@ class StallWatch(object):
                 continue
             backoff.record_success()
             fail_since = None
+            if not self._dispatch_open():
+                log.info("stall_watch: job %s closed or re-dispatched during the wait; stall watch stops", self.args.job)
+                return RC_TERMINAL
             if out.kind == watcher.MATCHED:
                 new = out.agent.get("agent_status")
                 sig.pane_id = out.agent.get("pane_id") or sig.pane_id
@@ -347,14 +381,29 @@ class StallWatch(object):
                 self._notify_once("never_empty", "stall_watch:input_never_empty:%s" % agent,
                                   "stall_watch: input box of %s never looked empty in %ds" % (agent, int(sig.idle_for)))
         if judge(sig, st) == NOTIFY_IDLE_AND_NUDGE:
+            if not self._dispatch_open():  # collect() took time; the job may have closed meanwhile
+                log.info("stall_watch: job %s closed or re-dispatched before the nudge; nothing sent", self.args.job)
+                return idle_since
             log.warning("stall_watch: agent %s idle for %ds with no background work; nudging once",
                         self.args.target, int(sig.idle_for))
-            self._notify("stalled:idle")
             st.nudged = True
+            if not self._notify("stalled:idle"):
+                log.warning("stall_watch: the stalled:idle notice was refused; nudge withheld")
+                return idle_since
             self.src.prompt(NUDGE.format(job=self.args.job))
         return idle_since
 
 
+def _alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    return True
+
+
 def main(argv=None) -> int:
     p = argparse.ArgumentParser(description="notify-only stall watcher for one local herdr job (#30)")
     p.add_argument("--job", required=True)
@@ -364,6 +413,7 @@ def main(argv=None) -> int:
     p.add_argument("--herdr-session", default="")
     p.add_argument("--expect-dispatch-token", default=None)
     p.add_argument("--attempt-id", type=int, default=None)
+    p.add_argument("--watch-pid", type=int, default=None, help="exit once this process (the wrapper) is gone")
     args = p.parse_args(argv)
     logging.basicConfig(level=logging.INFO, format="%(asctime)s stall_watch %(levelname)s %(message)s")
     # The wrapper stops us with SIGTERM once its subscriber ends: leave quietly.

.agents/skills/multi-agent-mux-delegate-job/SKILL.md

@@ -48,6 +48,8 @@ multi-agent-mux-delegate-job submit \
   #   argv string is capped at 131,072 bytes on Linux. The loop always uses it.
   #   MAM_JOB_TIMEOUT_S: the budget when --timeout is not given (default 3600).
   #   MAM_STALL_IDLE_S (600) / MAM_STALL_HERDR_FAIL_S (120) / MAM_STALL_WATCH=0: the stall watcher (Issue #30).
+  #   The watcher lives exactly as long as the job's wrapper: the wrapper stops it (TERM, 5 s, then KILL)
+  #   when its subscriber ends, and the watcher exits on its own once the wrapper pid is gone.
   #   --parent-job <id> (Issue #34): this job is delegated from job <id>, which
   #   must exist in the registry (else exit 1). A loop waiting on <id> waits for
   #   it too and shows it to reviewers. Default: $MAM_JOB_ID (the remote runner
@@ -96,9 +98,9 @@ Under fail-closed operation (`MAM_DELEGATE_FAIL_CLOSED=1`, default), the CLI ret
 |:---:|:---:|:---:|---|
 | `0` | Success | `completed` | Job succeeded and completed event received or verified on disk with approval (`PASS`/`AGREE`). |
 | `1` | Failure | `error` | Execution failure / delivery failure: no `started` within `--ack-timeout` T s after the injection (`agent_blocked`; the prompt is never re-injected), or the input never reached the pane. |
-| `2` | Timeout | `error` (`deadline_timeout`) | Subscriber timeout (total budget or idle timeout reached without terminal event). **R1 liveness grace (Issue #18)**: after the subscriber idles out the wrapper keeps re-observing the disk while the job's events.log shows an event at most K x N (`--idle-timeout`) old - the automatic heartbeat counts - up to the wall-clock `--timeout`. herdr status and the pane are logged as a diagnostic only. |
-| `2` | Wall clock | `error` (`wall_clock_timeout`) | The budget (`--timeout`) ran out, events or not (Issue #33; was `deadline_timeout` before v4.7.5). |
-| `3` | Broker Error | `error` (`deadline_timeout`) | Broker unavailable (`MQTT_BROKER_UNAVAILABLE`) and terminal status not found on disk during deadline. |
+| `2` | Timeout | `error` (`deadline_timeout`) | Subscriber idle timeout (K x N without an event; the total budget is the wall-clock row below). **R1 liveness grace (Issue #18)**: after the subscriber idles out the wrapper keeps re-observing the disk while the job's events.log shows an event at most K x N (`--idle-timeout`) old - the automatic heartbeat counts - up to the wall-clock `--timeout`. herdr status and the pane are logged as a diagnostic only. |
+| `2` | Wall clock | `error` (`wall_clock_timeout`) | The budget (`--timeout`) ran out, events or not (Issue #33; was `deadline_timeout` before v4.8.0). |
+| `3` | Broker Error | `error` (`wall_clock_timeout` once the budget is spent, else `deadline_timeout`) | Broker unavailable (`MQTT_BROKER_UNAVAILABLE`) and terminal status not found on disk during deadline. |
 | `5` | Session Busy | `cancelled` (`session_busy`) | Precondition failed / `session_busy`: Preflight check (`preflight_agent`) or TUI idle handshake wait failed (prompt never delivered, retryable). |
 
 In all execution paths, the final line of standard output prints the persistent audit log directory path (`<logs_root>/<job_id>`), fulfilling the parsing contract for orchestrator runners.

.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job

@@ -692,7 +692,7 @@ start_stall_watch() {
     echo "stall watch: human job, no local pane - stall detection off" >&2; return 0
   fi
   local args=(--job "$JOB_ID" --registry-dir "$REGISTRY_DIR" --target "${session#herdr:}"
-              --agent "$agent" --herdr-session "${HERDR_SESSION_NAME:-}")
+              --agent "$agent" --herdr-session "${HERDR_SESSION_NAME:-}" --watch-pid "$$")
   [ -n "$tok" ] && args+=(--expect-dispatch-token "$tok")
   [ -n "$att" ] && args+=(--attempt-id "$att")
   PYTHONPATH="$SCRIPT_DIR/..${PYTHONPATH:+:$PYTHONPATH}" "$PY" -m lib_py.stall_watch "${args[@]}" \
@@ -703,11 +703,23 @@ start_stall_watch() {
 
 # A watcher that already ended on its own with an error (herdr unavailable)
 # is not left silent: its last lines go to this wrapper's stderr.
+# The stop is bounded: a TERM that lands in the watcher's fork-exec window is
+# swallowed (bash defers terminating signals in the forked child until the
+# exec), and an unbounded `wait` then hung this wrapper until someone killed
+# the watcher by hand (PR #41 review N1). TERM, up to 5 s, then KILL.
 stop_stall_watch() {
   [ -n "${STALL_PID:-}" ] || return 0
-  local rc=0
+  local rc=0 i
   if kill -0 "$STALL_PID" 2>/dev/null; then
     kill "$STALL_PID" 2>/dev/null || true
+    for ((i = 0; i < 25; i++)); do
+      kill -0 "$STALL_PID" 2>/dev/null || break
+      sleep 0.2
+    done
+    if kill -0 "$STALL_PID" 2>/dev/null; then
+      echo "stall watch for $JOB_ID ignored TERM; killed (pid $STALL_PID)" >&2
+      kill -9 "$STALL_PID" 2>/dev/null || true
+    fi
     wait "$STALL_PID" 2>/dev/null || true
   else
     wait "$STALL_PID" 2>/dev/null || rc=$?

tests/test_issue30_stall_watch.py

@@ -36,7 +36,8 @@ from lib_py.agents.input_region import extract_input_region  # noqa: E402
 from lib_py import stall_watch  # noqa: E402
 from lib_py.loop import watcher  # noqa: E402
 
-pytestmark = pytest.mark.skipif(not sys.platform.startswith("linux"), reason="real process trees via /proc")
+# Only S6 needs Linux (/proc); the rest run on macOS too (PR #41 review M8).
+PROC = pytest.mark.skipif(not sys.platform.startswith("linux"), reason="real process trees via /proc")
 
 UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
 
@@ -327,6 +328,7 @@ def test_s5_the_adapters_resident_children_change_the_judgement(box, agent_name,
 # S6 - the heartbeat timer is never background work
 # ---------------------------------------------------------------------------
 
+@PROC
 def test_s6_the_heartbeat_timer_is_not_a_background_child(box):
     _, state, reg, env = box
     jid = _job(env, reg)
@@ -687,12 +689,14 @@ def test_r4_a_missed_transition_restarts_the_idle_window(mam_sandbox, mock_herdr
         #!/usr/bin/env python3
         import json, sys, time
         from pathlib import Path
+        # time.time(), not monotonic: these stamps are compared across processes,
+        # and macOS gives each process its own monotonic reference.
         f = Path(__file__).parent.parent / 'observed.json'
         state = json.loads(f.read_text())
         args = sys.argv[1:]
         if args[:2] == ['agent', 'wait']:
             if not state.get('started'):
-                state['started'] = time.monotonic()
+                state['started'] = time.time()
                 f.write_text(json.dumps(state))
                 print(json.dumps({'result': {'agent': {'agent_status': 'idle', 'state_change_seq': 0, 'pane_id': 'w1:p1'}}}))
             else:
@@ -701,7 +705,7 @@ def test_r4_a_missed_transition_restarts_the_idle_window(mam_sandbox, mock_herdr
                 sys.exit(1)
         elif args[:2] == ['agent', 'get']:
             if state['miss'] and not state.get('changed'):
-                state['changed'] = time.monotonic()
+                state['changed'] = time.time()
                 state['seq'] = 2
                 f.write_text(json.dumps(state))
             print(json.dumps({'result': {'agent': {'agent_status': 'idle', 'state_change_seq': state['seq'], 'pane_id': 'w1:p1'}}}))
@@ -712,7 +716,7 @@ def test_r4_a_missed_transition_restarts_the_idle_window(mam_sandbox, mock_herdr
         elif args[:2] == ['agent', 'read']:
             print(state['pane'])
         elif args[:2] == ['agent', 'prompt']:
-            state['prompted'] = time.monotonic()
+            state['prompted'] = time.time()
             f.write_text(json.dumps(state))
             print('{}')
         else:

tests/test_issue30_stall_watch_stop.py

@@ -0,0 +1,326 @@
+"""Issue #30 review (PR #41, N1): the wrapper never waits forever for its
+stall watcher, and the watcher never outlives its wrapper.
+
+The wrapper stops the watcher with SIGTERM the moment its subscriber ends.
+A TERM that lands inside the watcher's fork-exec window is swallowed (bash
+defers terminating signals in the forked child until the exec), so with an
+unbounded `wait` the wrapper hung until someone killed the watcher by hand.
+Both tests run on macOS and Linux: no /proc, no live broker.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+PUBLISH = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
+
+UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_HOST": "127.0.0.1", "MQTT_PORT": "65534", "MQTT_TLS": "0"}
+
+
+def _env(home: Path, **extra) -> dict:
+    env = {k: v for k, v in os.environ.items() if not k.startswith("HERDR_")}
+    env.update(UNREACHABLE, DELEGATE_JOB_PYTHON=sys.executable, AGENT_PYTHON_BIN=sys.executable,
+               PYTHONPATH=str(SKILLS), MAM_JOB_ID="", HOME=str(home), WORKSPACE_ROOT=str(home),
+               DELEGATE_JOB_LOGS_DIR=str(home / ".mam" / "logs"))
+    env.update({k: str(v) for k, v in extra.items()})
+    return env
+
+
+def _record(reg: Path, jid: str) -> dict:
+    return json.loads((reg / f"{jid}.json").read_text())
+
+
+def _publish(env: dict, reg: Path, jid: str, event: str, *extra: str) -> None:
+    res = subprocess.run([sys.executable, str(PUBLISH), "--registry-dir", str(reg), "--job", jid, "--event", event,
+                          "--expect-dispatch-token", _record(reg, jid)["dispatch_token"], "--attempts", "1", *extra],
+                         capture_output=True, text=True, env=env, timeout=120)
+    assert res.returncode in (0, 2), res.stderr
+
+
+def _alive(pid: int) -> bool:
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    return True
+
+
+def _until(pred, limit=20.0):
+    end = time.monotonic() + limit
+    while time.monotonic() < end:
+        if pred():
+            return True
+        time.sleep(0.2)
+    return pred()
+
+
+def test_stop_stall_watch_is_bounded_when_the_watcher_ignores_term(tmp_path):
+    """The wrapper's stop routine, run as is against a stand-in that ignores
+    SIGTERM (what a swallowed TERM looks like from the wrapper's side): it
+    returns within its 5 s budget, says so, and the stand-in is dead."""
+    fn = subprocess.run(["sed", "-n", "/^stop_stall_watch()/,/^}/p", str(WRAPPER)],
+                        capture_output=True, text=True, check=True).stdout
+    assert fn.startswith("stop_stall_watch()") and fn.rstrip().endswith("}")
+    script = (fn + '\nJOB_ID=t; REGISTRY_DIR="$1"\n'
+              '(trap "" TERM; exec sleep 300) & STALL_PID=$!\n'
+              'sleep 0.5; t0=$SECONDS; stop_stall_watch; echo "took=$((SECONDS - t0))"\n'
+              'kill -0 "$!" 2>/dev/null && echo "stand-in alive" || echo "stand-in gone"\n')
+    t0 = time.monotonic()
+    res = subprocess.run(["bash", "-c", script, "stop", str(tmp_path)], capture_output=True, text=True, timeout=30)
+    took = time.monotonic() - t0
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert "stand-in gone" in res.stdout, res.stdout
+    assert "stall watch for t ignored TERM; killed" in res.stderr, res.stderr
+    assert 4 < took < 12, took
+
+
+def test_the_wrapper_returns_when_its_watcher_ignores_term(tmp_path, mock_herdr):
+    """The Issue #14 ack scenario that hung: broker unreachable, the RPC fails,
+    the registry ack wins, the subscriber ends at once and the wrapper stops
+    the watcher. Here the watcher is a stand-in whose shim ignores SIGTERM from
+    its first line: the wrapper must still return, on its usual
+    broker_unavailable path, and leave no watcher behind."""
+    reg = tmp_path / "jobs"
+    reg.mkdir()
+    pidfile = tmp_path / "watcher.pid"
+    shim = tmp_path / "py-shim"
+    shim.write_text(
+        "#!/bin/sh\n"
+        'case "$*" in *lib_py.stall_watch*)\n'
+        "  trap '' TERM\n"
+        f'  exec "{sys.executable}" -c "import os, time; '
+        f"open({str(pidfile)!r}, 'w').write(str(os.getpid())); time.sleep(300)\" ;;\n"
+        "esac\n"
+        f'exec "{sys.executable}" "$@"\n')
+    shim.chmod(0o755)
+
+    state_file = Path(mock_herdr)
+    state = json.loads(state_file.read_text())
+    state["agents"]["ack-agent"] = {"agent": "codex", "agent_status": "idle", "prompt_wait_result": "transport_error"}
+    state["prompt_wait_result"] = "transport_error"
+    state_file.write_text(json.dumps(state, indent=2))
+
+    env = _env(tmp_path, DELEGATE_JOB_PYTHON=str(shim), SKS_IDLE_TIMEOUT_MS=500, MAM_INJECT_ACK_GRACE_S=8)
+    proc = subprocess.Popen(["bash", str(WRAPPER), "submit", "--agent", "codex", "--agent-session", "herdr:ack-agent",
+                             "--prompt", "issue 30 stop", "--registry-dir", str(reg), "--workdir", str(tmp_path),
+                             "--timeout", "6", "--idle-timeout", "2"],
+                            cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+    try:
+        assert _until(lambda: list(reg.glob("*.json")), 10), "job was not registered"
+        jid = list(reg.glob("*.json"))[0].stem
+        _publish(env, reg, jid, "started", "--attempt-id", "1")
+        t0 = time.monotonic()
+        stdout, stderr = proc.communicate(timeout=60)
+        took = time.monotonic() - t0
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+    assert proc.returncode == 3, f"expected the broker_unavailable path (rc=3), got {proc.returncode}\n{stderr}\n{stdout}"
+    # ack grace (8 s) + the 5 s TERM budget + slack; an unbounded wait never returns.
+    assert took < 40, took
+    if pidfile.exists():  # the stand-in got past its shim (a TERM in the shim's own start-up can still end it)
+        assert _until(lambda: not _alive(int(pidfile.read_text())), 5), "the watcher outlived the wrapper"
+        assert re.search(r"stall watch for \S+ ignored TERM", stderr), stderr
+
+
+def test_the_watcher_stops_once_its_wrapper_is_gone(mam_sandbox, mock_herdr):
+    """`--watch-pid`: a watcher whose wrapper vanished (killed, crashed) exits
+    with rc 0 on its own instead of polling herdr until the job closes."""
+    reg = mam_sandbox / ".mam" / "jobs"
+    env = _env(mam_sandbox, MAM_STALL_IDLE_S=600)
+    res = subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--agent-session", "herdr:img",
+                          "--dry-run", "--registry-dir", str(reg), "--prompt", "work"],
+                         capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
+    _publish(env, reg, jid, "started")
+    assert _record(reg, jid)["status"] == "running"
+
+    state_file = Path(mock_herdr)
+    state = json.loads(state_file.read_text())
+    state["wait_blocks_s"] = 0.5
+    state.setdefault("agents", {})["creator-codex-t"] = {
+        "name": "creator-codex-t", "agent": "codex", "status": "running", "agent_status": "idle",
+        "pane_id": "w1:p1", "workspace_id": "w1", "pid": 4242, "buffer": "› Ask Codex to do anything"}
+    state_file.write_text(json.dumps(state, indent=2))
+
+    stand_in = subprocess.Popen(["sleep", "300"])
+    w = subprocess.Popen([sys.executable, "-m", "lib_py.stall_watch", "--job", jid, "--registry-dir", str(reg),
+                          "--target", "creator-codex-t", "--agent", "codex", "--watch-pid", str(stand_in.pid)],
+                         env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    try:
+        time.sleep(3)
+        assert w.poll() is None, "the watcher left while its wrapper was alive:\n" + w.stdout.read()
+        stand_in.kill()
+        stand_in.wait()
+        out, _ = w.communicate(timeout=30)
+    finally:
+        if w.poll() is None:
+            w.kill()
+    assert w.returncode == 0, out
+    assert f"wrapper pid {stand_in.pid} is gone" in out, out
+    assert _record(reg, jid)["status"] == "running"  # notify-only: it closed nothing
+
+
+# ---------------------------------------------------------------------------
+# codex R1 - the real wrapper killed by TERM leaves no watcher behind
+# ---------------------------------------------------------------------------
+
+def test_a_terminated_wrapper_leaves_no_watcher_behind(mam_sandbox, mock_herdr):
+    """The wrapper's own functions: `start_stall_watch` then `wait`, and TERM to
+    the wrapper only. The orphaned watcher must end on its own (rc 0), since
+    no trap can run for a wrapper that is simply gone."""
+    reg = mam_sandbox / ".mam" / "jobs"
+    env = _env(mam_sandbox, MAM_STALL_IDLE_S=600)
+    res = subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--agent-session", "herdr:img",
+                          "--dry-run", "--registry-dir", str(reg), "--prompt", "work"],
+                         capture_output=True, text=True, env=env, cwd=str(mam_sandbox), timeout=120)
+    assert res.returncode == 0, res.stdout + res.stderr
+    jid = re.search(r"registered job:\s*(\S+)", res.stdout).group(1)
+    _publish(env, reg, jid, "started")
+
+    state_file = Path(mock_herdr)
+    state = json.loads(state_file.read_text())
+    state["wait_blocks_s"] = 0.5
+    state.setdefault("agents", {})["creator-codex-t"] = {
+        "name": "creator-codex-t", "agent": "codex", "status": "running", "agent_status": "idle",
+        "pane_id": "w1:p1", "workspace_id": "w1", "pid": 4242, "buffer": "› Ask Codex to do anything"}
+    state_file.write_text(json.dumps(state, indent=2))
+
+    script = ('MAM_DELEGATE_JOB_NO_MAIN=1 source "$1"; JOB_ID="$2"; REGISTRY_DIR="$3"; PY="$4"\n'
+              'start_stall_watch herdr:creator-codex-t codex\n'
+              'echo "WATCHER=$STALL_PID"; wait\n')
+    proc = subprocess.Popen(["bash", "-c", script, "probe", str(WRAPPER), jid, str(reg), sys.executable],
+                            env=env, cwd=str(mam_sandbox), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    wpid = None
+    try:
+        for line in proc.stdout:
+            if line.startswith("WATCHER="):
+                wpid = int(line.split("=", 1)[1])
+                break
+        assert wpid and _alive(wpid), "the watcher did not start"
+        time.sleep(2)
+        proc.terminate()  # TERM to the wrapper only
+        proc.wait(timeout=10)
+        assert _until(lambda: not _alive(wpid), 30), "the watcher outlived its wrapper"
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+        if wpid and _alive(wpid):
+            os.kill(wpid, 9)
+    out = (reg / f"{jid}.stall.out").read_text()
+    assert f"wrapper pid {proc.pid} is gone" in out, out
+    assert _record(reg, jid)["status"] == "running"
+
+
+# ---------------------------------------------------------------------------
+# codex R2 - a job that closed during a blocking wait is not nudged
+# ---------------------------------------------------------------------------
+
+class _Src:
+    """A source whose facts always say "idle, alone, empty input"."""
+
+    def __init__(self):
+        self.prompts = []
+
+    def collect(self, sig):
+        sig.state_change_seq, sig.agent_pid, sig.background_children, sig.input_empty = 1, 4242, [], True
+        return sig
+
+    def prompt(self, text):
+        self.prompts.append(text)
+
+    def blocked_text(self, pane_id):
+        return ""
+
+    def agent_get(self):
+        return {"pane_id": "w1:p1"}
+
+
+def _watch_in_process(tmp_path, monkeypatch, waits, notify):
+    from lib_py import stall_watch
+    from lib_py.loop import watcher as w
+    import argparse
+    import uuid
+    jid = uuid.uuid4().hex[:8]
+    rec = tmp_path / f"{jid}.json"
+    rec.write_text(json.dumps({"job_id": jid, "status": "running", "dispatch_token": "tok", "attempt_id": 1}))
+    calls = []
+
+    def fake_wait_once(target, until=(), timeout_ms=None, session=None, run=None):
+        calls.append(until)
+        return waits[min(len(calls), len(waits)) - 1](rec)
+
+    monkeypatch.setattr(stall_watch.watcher, "wait_once", fake_wait_once)
+    monkeypatch.setenv("MAM_STALL_IDLE_S", "1")
+    args = argparse.Namespace(job=jid, registry_dir=str(tmp_path), target="creator-codex-t", agent="codex",
+                              herdr_session="", expect_dispatch_token="tok", attempt_id=1, watch_pid=None)
+    src = _Src()
+    sw = stall_watch.StallWatch(args, source=src)
+    notes = []
+    monkeypatch.setattr(sw, "_notify", lambda detail: notes.append(detail) or notify)
+    rc = sw.run()
+    return rc, src.prompts, notes, rec, w
+
+
+def _idle(rec):
+    from lib_py.loop import watcher as w
+    return w.WaitOutcome(w.MATCHED, agent={"agent_status": "idle", "state_change_seq": 1, "pane_id": "w1:p1"})
+
+
+def _completed_during_the_wait(rec):
+    from lib_py.loop import watcher as w
+    time.sleep(1.2)  # past S while "blocked in herdr"
+    d = json.loads(rec.read_text())
+    d["status"] = "completed"
+    rec.write_text(json.dumps(d))
+    return w.WaitOutcome(w.TIMEOUT, error_code="timeout")
+
+
+def _redispatched_during_the_wait(rec):
+    from lib_py.loop import watcher as w
+    time.sleep(1.2)
+    d = json.loads(rec.read_text())
+    if d.get("ticks"):  # a watcher that ignores the new token is stopped by the close on the next tick
+        d["status"] = "completed"
+    d["ticks"] = d.get("ticks", 0) + 1
+    d["dispatch_token"], d["attempt_id"] = "tok-2", 2
+    rec.write_text(json.dumps(d))
+    return w.WaitOutcome(w.TIMEOUT, error_code="timeout")
+
+
+def test_a_job_completed_during_the_wait_is_not_nudged(tmp_path, monkeypatch):
+    rc, prompts, notes, _, _ = _watch_in_process(tmp_path, monkeypatch, [_idle, _completed_during_the_wait], True)
+    assert rc == 0 and prompts == [] and notes == [], (prompts, notes)
+
+
+def test_a_job_redispatched_during_the_wait_is_not_nudged(tmp_path, monkeypatch):
+    rc, prompts, notes, rec, _ = _watch_in_process(tmp_path, monkeypatch, [_idle, _redispatched_during_the_wait], True)
+    assert rc == 0 and prompts == [] and notes == [], (prompts, notes)
+    assert json.loads(rec.read_text())["status"] == "running"  # it left on the token, and closed nothing
+
+
+def test_a_refused_idle_notice_withholds_the_nudge(tmp_path, monkeypatch):
+    """The registry (CAS) or the wire refused the `stalled:idle` progress: no
+    prompt goes to the pane, and the job is then left alone (nudged once)."""
+    def _idle_then_close(rec):
+        from lib_py.loop import watcher as w
+        time.sleep(1.2)
+        d = json.loads(rec.read_text())
+        # the second idle tick nudges (or not); the third finds the job closed and ends the run
+        if d.get("ticks"):
+            d["status"] = "completed"
+        d["ticks"] = d.get("ticks", 0) + 1
+        rec.write_text(json.dumps(d))
+        return w.WaitOutcome(w.TIMEOUT, error_code="timeout")
+    rc, prompts, notes, _, _ = _watch_in_process(tmp_path, monkeypatch, [_idle, _idle_then_close], False)
+    assert rc == 0 and notes == ["stalled:idle"] and prompts == [], (prompts, notes)