From 9200562c914ab4f6e6d8a599cac371a07f99b25d Mon Sep 17 00:00:00 2001 From: Godopu Date: Sat, 26 Sep 2026 15:28:07 +0900 Subject: [PATCH] fix(issue-30): review round 2 - the watcher is stopped with KILL, and a strict last gate sits between the notice and the keys grok N1: the wrapper spent up to 5 s on a TERM budget before it finalized the job, so when the watcher swallowed the TERM the silent job's close came ~5 s late and a `completed` written in that gap won (test_issue14 idle-grace). codex D1: `_notify` blocks on a publish for up to 120 s, and the prompt went out right after it; a completed, a re-dispatch or an unreadable record in that window still got the nudge. codex D2: the wrapper pid was checked only at the loop head, so a watcher whose wrapper vanished during a wait could still notify and nudge once. - multi-agent-mux-delegate-job `stop_stall_watch`: KILL + wait, no budget. The watcher holds nothing to flush (its TERM handler is a bare os._exit), and KILL cannot be swallowed. - lib_py/stall_watch.py: `_wrapper_gone()` after every wait (before any notice) and at the loop head; `_may_prompt()` right before the keys - the wrapper is alive and the record, read then, is open and this dispatch; `_dispatch_open(strict=True)` treats an unreadable record as "no". - SKILL.md: the lifetime contract states the one-wait lag and that nothing is sent after the wrapper is seen gone. - tests/test_issue30_stall_watch_stop.py: the stop returns at once against a TERM-ignoring stand-in (took 6 s on the round-2 code); completed, re-dispatch and a torn record during the notice publish withhold the nudge; a wrapper gone during the wait gets no notice and no nudge. All five fail on the round-2 code. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff --- .agents/skills/lib_py/stall_watch.py | 35 ++++-- .../multi-agent-mux-delegate-job/SKILL.md | 6 +- .../multi-agent-mux-delegate-job | 21 ++-- tests/test_issue30_stall_watch_stop.py | 113 ++++++++++++++++-- 4 files changed, 144 insertions(+), 31 deletions(-) diff --git a/.agents/skills/lib_py/stall_watch.py b/.agents/skills/lib_py/stall_watch.py index b9500b1..ebee84f 100644 --- a/.agents/skills/lib_py/stall_watch.py +++ b/.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() diff --git a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md b/.agents/skills/multi-agent-mux-delegate-job/SKILL.md index acb11d7..c2fbefd 100644 --- a/.agents/skills/multi-agent-mux-delegate-job/SKILL.md +++ b/.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 (Issue #34): this job is delegated from job , which # must exist in the registry (else exit 1). A loop waiting on waits for # it too and shows it to reviewers. Default: $MAM_JOB_ID (the remote runner diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 172c28e..103501b 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.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=$? diff --git a/tests/test_issue30_stall_watch_stop.py b/tests/test_issue30_stall_watch_stop.py index a31da71..b2b9d5c 100644 --- a/tests/test_issue30_stall_watch_stop.py +++ b/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)