.agents/skills/lib_py/stall_watch.py

@@ -263,18 +263,20 @@ class StallWatch(object):
         except (OSError, ValueError):
             return None
 
-    def _dispatch_open(self):
+    def _dispatch_open(self, strict=False):
         """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."""
+        an unreadable one as a transient - unless `strict` (the check that
+        permits a prompt), where only a readable, open record of this dispatch
+        says yes (round-2 review D1)."""
         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
+            return not strict
         if rec.get("status") in TERMINAL_STATUSES:
             return False
         tok = self.args.expect_dispatch_token
@@ -285,6 +287,19 @@ class StallWatch(object):
             return False
         return True
 
+    def _wrapper_gone(self):
+        if self.args.watch_pid and not _alive(self.args.watch_pid):
+            log.info("stall_watch: wrapper pid %d is gone; stall watch stops", self.args.watch_pid)
+            return True
+        return False
+
+    def _may_prompt(self):
+        """The last gate before keys reach the pane: the wrapper is alive and the
+        record, read now, is this dispatch and open. Anything the notice's
+        publish (up to 120 s) let happen - a completed, a re-dispatch, a dead
+        wrapper, an unreadable record - withholds the nudge."""
+        return not self._wrapper_gone() and self._dispatch_open(strict=True)
+
     def run(self) -> int:
         st, sig, agent = self.st, Signals(), self.args.agent
         if not st.reliable:
@@ -295,11 +310,9 @@ class StallWatch(object):
             sig.job_status = self._job_status()
             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):
+            if self._wrapper_gone():
                 # 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)
+                # stop ourselves - here, after every wait and before a prompt.
                 return RC_TERMINAL
             if sig.agent_status is None:
                 until, tmo = watcher.VALID_STATES, 1000  # learn the current state
@@ -313,6 +326,8 @@ class StallWatch(object):
             else:
                 until, tmo = ("idle", "done", "blocked"), 60000
             out = watcher.wait_once(self.args.target, until=until, timeout_ms=tmo, session=self.args.herdr_session)
+            if self._wrapper_gone():  # round-2 review D2: before any notice
+                return RC_TERMINAL
             if out.kind in (watcher.TRANSPORT_ERROR, watcher.AGENT_NOT_FOUND):
                 code = "agent_not_found" if out.kind == watcher.AGENT_NOT_FOUND else (out.error_code or "transport_error")
                 now = time.monotonic()
@@ -390,6 +405,10 @@ class StallWatch(object):
             if not self._notify("stalled:idle"):
                 log.warning("stall_watch: the stalled:idle notice was refused; nudge withheld")
                 return idle_since
+            if not self._may_prompt():  # the publish can take long; re-read before the keys
+                log.info("stall_watch: job %s closed, re-dispatched or unreadable, or the wrapper left, "
+                         "during the notice; nudge withheld", self.args.job)
+                return idle_since
             self.src.prompt(NUDGE.format(job=self.args.job))
         return idle_since
 
@@ -416,7 +435,7 @@ def main(argv=None) -> int:
     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.
+    # The wrapper stops us with SIGKILL once its subscriber ends; a TERM from anyone else: leave quietly.
     signal.signal(signal.SIGTERM, lambda *_: os._exit(RC_TERMINAL))
     return StallWatch(args).run()
 

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

@@ -48,8 +48,10 @@ 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.
+  #   The watcher lives as long as the job's wrapper: the wrapper stops it (KILL, at once)
+  #   when its subscriber ends, and the watcher exits on its own once the wrapper pid is gone
+  #   (checked after every herdr wait, so it can outlive a killed wrapper by one wait, <= 60 s,
+  #   and never sends a notice or a nudge after it saw the wrapper 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

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

@@ -703,23 +703,16 @@ 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.
+# The stop is KILL, not TERM: a TERM can be swallowed (it hung this wrapper in
+# `wait`, PR #41 review N1), and a TERM budget in front of the job's close
+# clock delayed deadline_timeout (round-2 review). The watcher holds nothing
+# to flush - its TERM handler is a bare os._exit - so KILL loses nothing, and
+# `wait` returns at once.
 stop_stall_watch() {
   [ -n "${STALL_PID:-}" ] || return 0
-  local rc=0 i
+  local rc=0
   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
+    kill -9 "$STALL_PID" 2>/dev/null || true
     wait "$STALL_PID" 2>/dev/null || true
   else
     wait "$STALL_PID" 2>/dev/null || rc=$?

tests/test_issue30_stall_watch_stop.py

@@ -63,10 +63,10 @@ def _until(pred, limit=20.0):
     return pred()
 
 
-def test_stop_stall_watch_is_bounded_when_the_watcher_ignores_term(tmp_path):
+def test_stop_stall_watch_returns_at_once_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."""
+    returns at once 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("}")
@@ -79,8 +79,8 @@ def test_stop_stall_watch_is_bounded_when_the_watcher_ignores_term(tmp_path):
     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
+    assert "took=0" in res.stdout or "took=1" in res.stdout, res.stdout
+    assert took < 5, took
 
 
 def test_the_wrapper_returns_when_its_watcher_ignores_term(tmp_path, mock_herdr):
@@ -125,11 +125,10 @@ def test_the_wrapper_returns_when_its_watcher_ignores_term(tmp_path, mock_herdr)
         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.
+    # ack grace (8 s) + 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)
+    if pidfile.exists():  # the stand-in got past its shim
         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):
@@ -324,3 +323,103 @@ def test_a_refused_idle_notice_withholds_the_nudge(tmp_path, monkeypatch):
         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)
+
+
+# ---------------------------------------------------------------------------
+# round-2 review (codex D1/D2) - the last gate sits between the notice and the keys
+# ---------------------------------------------------------------------------
+
+def _during_notice(tmp_path, monkeypatch, change, alive=lambda pid: True):
+    """Idle past S on the second tick; `change(rec)` runs inside the notice's
+    publish, as another actor would act while it waits on the wire."""
+    from lib_py import stall_watch
+    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}))
+    ticks = []
+
+    def fake_wait_once(target, until=(), timeout_ms=None, session=None, run=None):
+        from lib_py.loop import watcher as w
+        ticks.append(1)
+        if len(ticks) == 1:
+            return w.WaitOutcome(w.MATCHED, agent={"agent_status": "idle", "state_change_seq": 1, "pane_id": "w1:p1"})
+        if len(ticks) > 2:  # a watcher that did not stop: end the run through the record
+            d = json.loads(rec.read_text())
+            d["status"] = "completed"
+            rec.write_text(json.dumps(d))
+        time.sleep(1.2)
+        return w.WaitOutcome(w.TIMEOUT, error_code="timeout")
+
+    monkeypatch.setattr(stall_watch.watcher, "wait_once", fake_wait_once)
+    monkeypatch.setattr(stall_watch, "_alive", alive)
+    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=4242)
+    src = _Src()
+    sw = stall_watch.StallWatch(args, source=src)
+    notes = []
+
+    def notify(detail):
+        notes.append(detail)
+        change(rec)
+        return True  # the publisher said rc 0: that is not a statement about the record
+
+    monkeypatch.setattr(sw, "_notify", notify)
+    return sw.run(), src.prompts, notes
+
+
+def _complete(rec):
+    d = json.loads(rec.read_text())
+    d["status"] = "completed"
+    rec.write_text(json.dumps(d))
+
+
+def _redispatch(rec):
+    d = json.loads(rec.read_text())
+    d["dispatch_token"], d["attempt_id"] = "tok-2", 2
+    rec.write_text(json.dumps(d))
+
+
+def test_a_job_completed_while_the_notice_is_published_is_not_nudged(tmp_path, monkeypatch):
+    rc, prompts, notes = _during_notice(tmp_path, monkeypatch, _complete)
+    assert rc == 0 and notes == ["stalled:idle"] and prompts == [], (prompts, notes)
+
+
+def test_a_job_redispatched_while_the_notice_is_published_is_not_nudged(tmp_path, monkeypatch):
+    rc, prompts, notes = _during_notice(tmp_path, monkeypatch, _redispatch)
+    assert rc == 0 and notes == ["stalled:idle"] and prompts == [], (prompts, notes)
+
+
+def test_an_unreadable_record_at_the_last_gate_withholds_the_nudge(tmp_path, monkeypatch):
+    real = {}
+
+    def garble(rec):
+        real["text"] = rec.read_text()
+        rec.write_text("{not json")  # a torn write, seen once
+
+    from lib_py import stall_watch
+    orig = stall_watch.StallWatch._dispatch_open
+
+    def once_then_restore(self, strict=False):
+        out = orig(self, strict)
+        if strict and "text" in real:
+            (Path(self.args.registry_dir) / f"{self.args.job}.json").write_text(real.pop("text"))
+        return out
+
+    monkeypatch.setattr(stall_watch.StallWatch, "_dispatch_open", once_then_restore)
+    rc, prompts, notes = _during_notice(tmp_path, monkeypatch, garble)
+    assert rc == 0 and notes == ["stalled:idle"] and prompts == [], (prompts, notes)
+
+
+def test_a_wrapper_gone_during_the_wait_gets_no_notice_and_no_nudge(tmp_path, monkeypatch):
+    """Idle past S and the wrapper vanished during the same wait: the watcher
+    leaves before judging - no notice, no keys, the record untouched."""
+    calls = []
+
+    def alive(pid):
+        calls.append(pid)
+        return len(calls) <= 2  # alive at the first loop head and after the first wait, gone after
+    rc, prompts, notes = _during_notice(tmp_path, monkeypatch, lambda rec: None, alive=alive)
+    assert rc == 0 and notes == [] and prompts == [], (prompts, notes)