From bf0f5c101417d262805ca405b8343041f6aaba7b Mon Sep 17 00:00:00 2001 From: Godopu Date: Fri, 25 Sep 2026 14:01:53 +0000 Subject: [PATCH] fix(issue-30-33): review 5ec8ef33 - a multi-line codex draft is not an empty box, blocked notices reach the loop while the job runs, and both clocks are labelled on every close path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - R1 (P1): without rules the input region runs from the last prompt line to the end of the pane; only a last line matching the adapter's footer pattern (codex: ` · ~/ · …`) is dropped, so a draft on a second line keeps the nudge withheld. - R2: while the synchronous submit runs, a relay thread finds the job it registers (same session, created since the submit) and prints its stalled:/stall_watch: notices; before, they appeared only after the job. - R3: the broker-unavailable disk wait that runs into the deadline is the wall clock: wall_clock_timeout (caller rc stays 3). - R4: a state_change_seq change seen right before the judgement restarts the idle window (decision 9: S seconds of continuous idle). Regression tests are the reviewer's reproductions, each with its passing twin; the wrapper-level #30 tests now use this environment's broker instead of the #18 harness's public default. Co-Authored-By: Claude Opus 5.5 --- .../skills/lib_py/agents/adapters/codex.py | 6 + .agents/skills/lib_py/agents/base.py | 4 +- .agents/skills/lib_py/agents/input_region.py | 19 +- .agents/skills/lib_py/loop/runner.py | 27 +++ .agents/skills/lib_py/stall_watch.py | 18 +- .../multi-agent-mux-delegate-job | 2 + tests/test_issue30_stall_watch.py | 186 +++++++++++++++++- tests/test_issue33_budget.py | 19 ++ 8 files changed, 269 insertions(+), 12 deletions(-) diff --git a/.agents/skills/lib_py/agents/adapters/codex.py b/.agents/skills/lib_py/agents/adapters/codex.py index 183b7d1..b26d719 100644 --- a/.agents/skills/lib_py/agents/adapters/codex.py +++ b/.agents/skills/lib_py/agents/adapters/codex.py @@ -51,6 +51,12 @@ class CodexAgentAdapter(BaseAgentAdapter): def input_placeholder(self) -> Optional[str]: return 'Ask Codex to do anything' + @property + def input_stop_pattern(self) -> Optional[str]: + # The status footer under the input line: ` · · …` + # (the ` · ~/` or ` · /` of the cwd keeps a draft line with dots from matching) + return r'^\s+\S.*\s\u00b7\s[~/]' + @property def resident_children(self) -> tuple: # Kept alive while idle (measured 2026-09-25). diff --git a/.agents/skills/lib_py/agents/base.py b/.agents/skills/lib_py/agents/base.py index 6e3ca1e..0fd62a6 100644 --- a/.agents/skills/lib_py/agents/base.py +++ b/.agents/skills/lib_py/agents/base.py @@ -164,7 +164,9 @@ class BaseAgentAdapter: @property def input_stop_pattern(self) -> Optional[str]: - """Regex of the status line that ends the input area inside such a block (#30).""" + """Regex of the status line that ends the input area (#30): inside a + block, the first matching line; below a prompt line (no rules), only + the last non-blank line of the pane.""" return None @property diff --git a/.agents/skills/lib_py/agents/input_region.py b/.agents/skills/lib_py/agents/input_region.py index 939fdde..55e16ce 100644 --- a/.agents/skills/lib_py/agents/input_region.py +++ b/.agents/skills/lib_py/agents/input_region.py @@ -11,7 +11,10 @@ def extract_input_region(pane_text: str, agent_name: str) -> str: 1. an `input_block_prefix` -> the last contiguous block of lines starting with it, prefix removed, cut at the first `input_stop_pattern` line; 2. two or more `input_rule_pattern` rules -> the lines between the last two; - 3. an `input_prompt` -> the last line starting with the prompt. + 3. an `input_prompt` -> from the last line starting with the prompt to + the end of the pane, minus a last non-blank line matching + `input_stop_pattern` (the status footer). Every other line below the + prompt line is input: a multi-line draft is never cut to its first line. Then `input_frame_chars`, the prompt and the placeholder are stripped. There is no whole-pane fallback: a pane with none of these shapes has no input @@ -49,9 +52,17 @@ def extract_input_region(pane_text: str, agent_name: str) -> str: region_lines = lines[rule_indices[-2] + 1:rule_indices[-1]] if region_lines is None: - starts = [line for line in lines if line.lstrip().startswith(prompt)] - # No input line at all: report the whole tail as "not empty" (unknown). - region_lines = starts[-1:] if starts else [pane_text or '?'] + starts = [i for i, line in enumerate(lines) if line.lstrip().startswith(prompt)] + if starts: + region_lines = [line for line in lines[starts[-1]:]] + while region_lines and not region_lines[-1].strip(): + region_lines.pop() + if (len(region_lines) > 1 and adapter.input_stop_pattern + and re.search(adapter.input_stop_pattern, region_lines[-1])): + region_lines.pop() + else: + # No input line at all: report the whole tail as "not empty" (unknown). + region_lines = [pane_text or '?'] frame = adapter.input_frame_chars if frame: diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 7fee63d..cbce7b7 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -20,6 +20,7 @@ import sqlite3 import subprocess import sys import tempfile +import threading import time from pathlib import Path from typing import Any, Dict, List, Optional, Set, Tuple @@ -534,6 +535,11 @@ class LoopRunner: env = dict(os.environ) if "DELEGATE_JOB_PYTHON" not in env: env["DELEGATE_JOB_PYTHON"] = sys.executable + stop = threading.Event() + relay = threading.Thread(target=self._relay_while_submitting, + args=(self.session_label(agent_session), job_registry._utcnow(), stop), + daemon=True) + relay.start() try: return subprocess.run(cmd, capture_output=True, text=True, cwd=str(self.workspace_root), env=env) except OSError as e: @@ -543,6 +549,8 @@ class LoopRunner: f"(prompt {len(p_text.encode('utf-8'))} bytes, wrapper {wrapper_script})") return subprocess.CompletedProcess(cmd, 126, "", msg) finally: + stop.set() + relay.join(timeout=5) try: os.unlink(prompt_file) except OSError: @@ -652,6 +660,25 @@ class LoopRunner: log_error(f"Job '{job_id}' timed out after {max_wait}s.") return False + def _relay_while_submitting(self, label: str, since: str, stop: threading.Event) -> None: + """Issue #30 (review R2): the wrapper runs until its job ends and its + output is captured, so a blocked agent would only reach the loop log + after the job. While the submit runs, find the job it registers (this + session, created since the submit started) and relay its notices.""" + jid = None + while not stop.wait(1.0): + if jid is None: + try: + jobs = job_registry.list_jobs(str(self.jobs_dir)) + except (OSError, ValueError): + continue + mine = [r for r in jobs if r.get("agent_session") == label + and str(r.get("created_at") or "") >= since] + if not mine: + continue + jid = str(mine[-1].get("job_id")) + self._relay_stall_warnings(jid) + def _relay_stall_warnings(self, job_id: str) -> None: """Issue #30: surface the stall watcher's progress events (detail `stalled:...` / `stall_watch:...`) in the loop log, once per detail. diff --git a/.agents/skills/lib_py/stall_watch.py b/.agents/skills/lib_py/stall_watch.py index 7008b8a..4142b04 100644 --- a/.agents/skills/lib_py/stall_watch.py +++ b/.agents/skills/lib_py/stall_watch.py @@ -307,11 +307,13 @@ class StallWatch(object): st.blocked_notified = False sig.agent_status = new try: - self._act(sig, idle_since) + idle_since = self._act(sig, idle_since) except HerdrError as exc: log.warning("stall_watch: herdr call failed while judging: %s", exc.code) def _act(self, sig, idle_since): + """Notify/nudge per `judge`; returns the idle start to keep (restarted + when a transition was missed between the wait and the judgement).""" st, agent = self.st, self.args.agent if sig.agent_status == "blocked" and not st.blocked_notified: sig.blocked_text = self.src.blocked_text(sig.pane_id or self.src.agent_get().get("pane_id", "")) @@ -319,16 +321,21 @@ class StallWatch(object): log.warning("stall_watch: agent %s is blocked: %s", self.args.target, sig.blocked_text) self._notify(("stalled:blocked " + sig.blocked_text).strip()) st.blocked_notified = True - return + return idle_since if sig.agent_status not in IDLE_STATES or not st.reliable or st.nudged: - return + return idle_since sig.idle_for = time.monotonic() - idle_since if sig.idle_for < st.idle_s: - return + return idle_since seq = sig.state_change_seq self.src.collect(sig) if seq is not None and sig.state_change_seq is not None and sig.state_change_seq != seq: - return # a transition we did not see: the next wait re-observes + # Review R4: a transition we did not see. Decision 9 wants S seconds + # of *continuous* idle, so the window restarts from this observation + # (collect() already took the new seq as the baseline). + log.info("stall_watch: state_change_seq %s -> %s since the idle start; idle window restarts", + seq, sig.state_change_seq) + return time.monotonic() if sig.background_children: log.debug("stall_watch: idle but background children %s", sig.background_children) elif sig.input_empty is None: @@ -345,6 +352,7 @@ class StallWatch(object): self._notify("stalled:idle") st.nudged = True self.src.prompt(NUDGE.format(job=self.args.job)) + return idle_since def main(argv=None) -> int: 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 70780cc..a798d69 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 @@ -896,6 +896,8 @@ finalize_unfinished() { break fi done + # Issue #33 (review R3): a wait that ran into the deadline is the wall clock. + [ "$(date +%s)" -ge "$deadline" ] && FINALIZE_CLOCK="wall" fi # timeout: if subscriber timed out, check if agent is still actively working in herdr diff --git a/tests/test_issue30_stall_watch.py b/tests/test_issue30_stall_watch.py index a1b6d4b..0072e3a 100644 --- a/tests/test_issue30_stall_watch.py +++ b/tests/test_issue30_stall_watch.py @@ -486,6 +486,14 @@ def test_scoped_env_drops_the_pane_inherited_workspace_for_name_lookups_only(): import test_issue18_liveness_contract as t18 # noqa: E402 +LIVE_BROKER = ("MQTT_BROKER", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD") + + +def _live_broker() -> dict: + """This environment's broker (the #18 harness defaults to a public one).""" + return {k: os.environ[k] for k in LIVE_BROKER if k in os.environ} + + def _watchers(jid: str) -> list: out = subprocess.run(["pgrep", "-f", f"lib_py.stall_watch --job {jid}"], capture_output=True, text=True).stdout return [int(p) for p in out.split()] @@ -497,7 +505,7 @@ def test_s9_the_wrapper_runs_the_watcher_for_the_jobs_life(tmp_path, mock_herdr, reg.mkdir() t18._headless_agent(Path(mock_herdr), "headless-codex") proc, env = t18._submit(tmp_path, reg, "herdr:headless-codex", "codex", ["--ack-timeout", "40"], - env_extra={"MAM_STALL_WATCH": switch}, timeout_s=60) + env_extra={"MAM_STALL_WATCH": switch, **_live_broker()}, timeout_s=60) jid = t18._wait_job_id(reg) try: time.sleep(1) @@ -523,7 +531,7 @@ def test_s10_remote_and_human_jobs_are_not_watched(tmp_path, agent, session, why reg = tmp_path / "jobs" reg.mkdir() proc, env = t18._submit(tmp_path, reg, session, agent, ["--ack-timeout", "40"], timeout_s=60, - env_extra={"MAM_DISPATCH_SECRET": "s10-secret"}) + env_extra={"MAM_DISPATCH_SECRET": "s10-secret", **_live_broker()}) jid = t18._wait_job_id(reg) try: time.sleep(1) @@ -563,3 +571,177 @@ def test_s11_the_runner_warns_once_per_stall_notice(mam_sandbox, monkeypatch, ca assert text.count(f"job {jid}: stalled:idle") == 1, text assert text.count(f"job {jid}: stall_watch:herdr_unavailable:protocol_mismatch") == 1, text assert "ordinary" not in text + + +# --------------------------------------------------------------------------- +# Review 5ec8ef33 (planner-reviewer-codex-01) regressions, taken from the +# reviewer's independent reproductions (codex-reports/test_review_regressions.py) +# R1 multi-line codex draft, R2 live relay during the synchronous submit, +# R4 continuous idle after a missed transition. Each has its passing twin. +# --------------------------------------------------------------------------- + +import textwrap # noqa: E402 + +@pytest.mark.parametrize("typing", [False, True]) +def test_r1_a_multi_line_codex_draft_is_not_an_empty_box(mam_sandbox, mock_herdr, typing): + reg = mam_sandbox / ".mam" / "jobs" + env = _env(mam_sandbox, MAM_STALL_IDLE_S=1) + jid = _job(env, reg) + agent = _proc("exec sleep 300") + w = None + try: + pane = _pane("codex") + if typing: + pane = pane.replace("› Ask Codex to do anything", "›\n draft on the second input line") + _agent(mock_herdr, "reviewer-codex-t", "codex", agent.pid, "idle", pane) + w = _watch(env, reg, jid, "reviewer-codex-t", "codex") + if typing: + time.sleep(5) + else: + assert _until(lambda: _prompts(mock_herdr)) + prompts = _prompts(mock_herdr) + _finish(env, reg, jid, w) + assert len(prompts) == (0 if typing else 1), prompts + finally: + if w is not None and w.poll() is None: + w.terminate() + w.communicate(timeout=10) + _kill(agent) + + +@pytest.mark.parametrize("blocked", [False, True]) +def test_r2_the_loop_relays_a_blocked_notice_while_the_job_is_open(mam_sandbox, mock_herdr, blocked): + reg = mam_sandbox / ".mam" / "jobs" + reg.mkdir(parents=True, exist_ok=True) + t18._headless_agent(mock_herdr, "headless-codex") + env = t18._env(mam_sandbox, {"MAM_STALL_WATCH": "1", "MAM_STALL_IDLE_S": "600", + "SKS_IDLE_TIMEOUT_MS": "500", "SKS_EMPTY_GIVEUP": "1", + "MAM_ACK_TIMEOUT_S": "40", "MAM_JOBS_DIR": str(reg), + "PYTHONPATH": str(SKILLS), "PYTHONUNBUFFERED": "1"}) + env.update(_live_broker()) + script = '''from lib_py.loop.runner import LoopRunner +r = LoopRunner(["--creator", "headless-codex", "--task", "review check", "--job-timeout", "60"]) +r.parse_args() +jid = r.delegate_job("headless-codex", "codex", "Worker", "review check") +assert jid +assert r.wait_for_job(jid, max_wait=10) +''' + p = subprocess.Popen([sys.executable, "-c", script], cwd=mam_sandbox, env=env, + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + jid = None + try: + jid = t18._wait_job_id(reg) + time.sleep(1) + assert t18._publish(reg, jid, "started", env).returncode == 0 + assert _until(lambda: (reg / f"{jid}.stall.out").exists()) + state = json.loads(mock_herdr.read_text()) + state["wait_blocks_s"] = 0.3 + state["agents"]["headless-codex"].update( + agent_status="blocked" if blocked else "working", pane_id="w1:p1", + detection="Approval needed. Do you want to proceed?") + mock_herdr.write_text(json.dumps(state)) + if blocked: + assert _until(lambda: any(d.startswith("stalled:blocked") for d in _details(reg, jid)), 40) + time.sleep(1) + assert _record(reg, jid)["status"] == "running" + os.set_blocking(p.stdout.fileno(), False) + live = bytearray() + while True: + try: + chunk = os.read(p.stdout.fileno(), 65536) + except BlockingIOError: + break + if not chunk: + break + live.extend(chunk) + os.set_blocking(p.stdout.fileno(), True) + assert t18._publish(reg, jid, "completed", env).returncode == 0 + tail, _ = p.communicate(timeout=60) + assert p.returncode == 0, (bytes(live) + tail).decode(errors="replace") + assert (b"stalled:blocked" in live) is blocked, { + "while_running": bytes(live).decode(errors="replace"), + "after_completion": tail.decode(errors="replace"), + } + finally: + if p.poll() is None: + p.kill() + p.communicate(timeout=10) + + +@pytest.mark.parametrize("missed_transition", [False, True]) +def test_r4_a_missed_transition_restarts_the_idle_window(mam_sandbox, mock_herdr, missed_transition): + """Real watcher/clock/process tree; an external herdr stub models a missed transition. + + The wait/get agent snapshots include state_change_seq, unlike the repository's + wait mock. process-info retains its measured nested result shape. + """ + reg = mam_sandbox / ".mam" / "jobs" + env = _env(mam_sandbox, MAM_STALL_IDLE_S=10) + jid = _job(env, reg) + agent = _proc("exec sleep 300") + observed = mam_sandbox / "observed.json" + observed.write_text(json.dumps({"seq": 0, "miss": missed_transition, "pid": agent.pid, + "pane": _pane("codex")})) + shim = mam_sandbox / "bin" / "herdr" + shim.write_text(textwrap.dedent('''\ + #!/usr/bin/env python3 + import json, sys, time + from pathlib import Path + 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() + f.write_text(json.dumps(state)) + print(json.dumps({'result': {'agent': {'agent_status': 'idle', 'state_change_seq': 0, 'pane_id': 'w1:p1'}}})) + else: + time.sleep(int(args[args.index('--timeout') + 1]) / 1000) + print(json.dumps({'error': {'code': 'timeout'}}), file=sys.stderr) + sys.exit(1) + elif args[:2] == ['agent', 'get']: + if state['miss'] and not state.get('changed'): + state['changed'] = time.monotonic() + 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'}}})) + elif args[:2] == ['pane', 'process-info']: + print(json.dumps({'result': {'type': 'pane_process_info', 'process_info': { + 'pane_id': 'w1:p1', 'shell_pid': 1, 'foreground_process_group_id': state['pid'], + 'foreground_processes': [{'pid': state['pid'], 'name': 'codex', 'argv': ['codex'], 'cmdline': 'codex', 'cwd': '/tmp'}]}}})) + elif args[:2] == ['agent', 'read']: + print(state['pane']) + elif args[:2] == ['agent', 'prompt']: + state['prompted'] = time.monotonic() + f.write_text(json.dumps(state)) + print('{}') + else: + sys.exit(2) + ''')) + shim.chmod(0o755) + w = None + try: + w = _watch(env, reg, jid, "reviewer-codex-t", "codex") + assert _until(lambda: 'prompted' in json.loads(observed.read_text()), 90) + times = json.loads(observed.read_text()) + # It may still be in a 30-second herdr wait after sending the prompt. + _publish(env, reg, jid, 'completed') + w.terminate() + w.communicate(timeout=10) + idle_time = times['prompted'] - times.get('changed', times['started']) + assert idle_time >= 10, {'continuous_idle_seconds': idle_time, 'threshold': 10} + finally: + if w is not None and w.poll() is None: + w.terminate() + w.communicate(timeout=10) + _kill(agent) + + +def test_r1_the_codex_extractor_keeps_every_line_under_the_prompt(): + from lib_py.agents.input_region import extract_input_region + idle = _pane("codex") + assert extract_input_region(idle, "codex") == "" + draft = idle.replace("› Ask Codex to do anything", "›\n draft on the second input line") + assert extract_input_region(draft, "codex") == "draft on the second input line" + no_footer = "\n".join(idle.splitlines()[:-1]).replace("› Ask Codex to do anything", "›\n a · b · c") + assert extract_input_region(no_footer, "codex") != "", "only the last line may be the footer" diff --git a/tests/test_issue33_budget.py b/tests/test_issue33_budget.py index 996e86a..5965080 100644 --- a/tests/test_issue33_budget.py +++ b/tests/test_issue33_budget.py @@ -322,3 +322,22 @@ def test_b5_a_wall_clock_close_is_closed_unconfirmed(tmp_path, monkeypatch, requ _publish(env, reg, c, "error", "--detail", detail) block = _runner(tmp_path, monkeypatch)._subjobs_block([p]) assert f"| CLOSED-UNCONFIRMED | {detail} |" in block, block + + +# Review 5ec8ef33 R3: the broker-unavailable wait that runs into the deadline +# is the wall clock too (caller rc stays 3); broker_up is the passing twin. +@pytest.mark.parametrize("broker_up", [True, False]) +def test_r3_a_budget_spent_waiting_on_an_unreachable_broker_is_wall_clock_timeout(tmp_path, broker_up): + env = _env(tmp_path, broker=broker_up) + reg = tmp_path / "jobs" + p = _run_wrapper_bg(env, reg, "--timeout", "4", "--idle-timeout", "30") + try: + out, err = p.communicate(timeout=90) + jid = next(reg.glob("*.json")).stem + assert p.returncode == (2 if broker_up else 3), out + err + assert _record(reg, jid)["status"] == "error" + assert _last_error(reg, jid) == "wall_clock_timeout", err + finally: + if p.poll() is None: + p.kill() + p.communicate(timeout=10)