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

- 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: `  <model> · ~/<cwd> · …`) 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 <noreply@anthropic.com>

.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: `  <model> <effort> · <cwd> · …`
+        # (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).

.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

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

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

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

.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

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"

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)