fix(issue-33): subscriber exit codes are named on both sides; wait keeps a spent budget at rc 2; G-8 is deterministic

- job_subscriber.py returns only named codes (RC_COMPLETED 0, RC_ERROR 1,
  RC_NO_EVENT 2, RC_BROKER_UNAVAILABLE 3, RC_WALL_CLOCK 4). The wrapper
  mirrors them as SUB_RC_* and compares subscriber codes only by name; a test
  keeps the mirror equal to the Python constants and forbids numeric
  comparisons of $sub_rc/$sub_exit and numeric `return` in the subscriber.
- `delegate-job wait` passed the subscriber's code straight through, so a
  spent budget would have reached its callers as 4; it is 2 again (B7 fails
  without the mapping).
- tier1 G-8 gave both clocks 0.5 s and raced; the wall clock now runs out
  first by construction (idle 30 s) and the test asserts RC_WALL_CLOCK.
  Other subscriber rc asserts use the names.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.agents/skills/lib_py/agents/__main__.py

@@ -59,11 +59,16 @@ def main():
         print(f"MAM_INPUT_PROMPT={q(adapter.input_prompt or '')}")
         print(f"MAM_INPUT_PLACEHOLDER={q(adapter.input_placeholder or '')}")
         print(f"MAM_INPUT_RULE_PATTERN={q(adapter.input_rule_pattern or '')}")
+        print(f"MAM_INPUT_FRAME_CHARS={q(adapter.input_frame_chars)}")
+        print(f"MAM_INPUT_BLOCK_PREFIX={q(adapter.input_block_prefix or '')}")
+        print(f"MAM_INPUT_STOP_PATTERN={q(adapter.input_stop_pattern or '')}")
         print(f"MAM_READY_TOKENS={q(adapter.ready_tokens)}")
         print(f"MAM_STRONG_READY_TOKENS={q(adapter.strong_ready_tokens)}")
         print(f"MAM_WEAK_READY_TOKENS={q(adapter.weak_ready_tokens)}")
         print(f"MAM_MODAL_TOKENS={q(adapter.modal_tokens or '')}")
         print(f"MAM_STARTUP_DIALOGS={q(json.dumps(adapter.startup_dialogs))}")
+        print(f"MAM_RESIDENT_CHILDREN={q(json.dumps(list(adapter.resident_children)))}")
+        print(f"MAM_AGENT_STATUS_RELIABLE={1 if adapter.agent_status_reliable else 0}")
         print(f"MAM_EXIT_KEY={q(adapter.exit_key)}")
         print(f"MAM_DELEGATE_AGENT_KEY={q(adapter.delegate_agent_key)}")
 

.agents/skills/lib_py/agents/adapters/codex.py

@@ -42,10 +42,26 @@ class CodexAgentAdapter(BaseAgentAdapter):
     def identity_cache_fields(self) -> tuple:
         return ('session_id',)
 
+    @property
+    def input_prompt(self) -> Optional[str]:
+        # herdr 0.9.1 capture: one `› Ask Codex to do anything` line, no rules.
+        return '›'
+
     @property
     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).
+        return ('codex-code-mode-host',)
+
     def _codex_home(self, ctx: DiscoveryContext) -> str:
         return os.environ.get('CODEX_HOME') or f"{ctx.home_dir}/.codex"
 

.agents/skills/lib_py/agents/adapters/grok.py

@@ -49,6 +49,11 @@ class GrokAgentAdapter(BaseAgentAdapter):
     def input_rule_pattern(self) -> str:
         return '─{10,}'
 
+    @property
+    def input_frame_chars(self) -> str:
+        # The input box is drawn as ╭─╮ / │ ❯ │ / ╰─╯ (herdr 0.9.1 capture).
+        return '│'
+
     def _ws_dir(self, ctx: DiscoveryContext) -> str:
         return quote(os.path.realpath(ctx.cwd), safe='')
 

.agents/skills/lib_py/agents/adapters/opencode.py

@@ -45,8 +45,20 @@ class OpenCodeAgentAdapter(BaseAgentAdapter):
 
     @property
     def input_rule_pattern(self) -> Optional[str]:
+        # Kept for lib.sh's readiness corroboration; the input box itself is a
+        # ┃ block (input_block_prefix), which the extractor reads first.
         return '─{10,}'
 
+    @property
+    def input_block_prefix(self) -> Optional[str]:
+        return '┃'
+
+    @property
+    def input_stop_pattern(self) -> Optional[str]:
+        # The input area ends at the right-aligned cwd line (a long run of
+        # blanks after the ┃) or the `Build auto · <model>` status line.
+        return r'^\s{20,}\S|^\s*(Build|Plan)\b.*·'
+
     def _resolve_db(self, ctx: DiscoveryContext) -> Optional[str]:
         # 1. Try opencode CLI "opencode db path"
         try:

.agents/skills/lib_py/agents/base.py

@@ -152,6 +152,36 @@ class BaseAgentAdapter:
     def input_rule_pattern(self) -> Optional[str]:
         return None
 
+    @property
+    def input_frame_chars(self) -> str:
+        """Box-border characters stripped from both ends of an input line (#30)."""
+        return ''
+
+    @property
+    def input_block_prefix(self) -> Optional[str]:
+        """Line prefix of an input box drawn as a block instead of between rules (#30)."""
+        return None
+
+    @property
+    def input_stop_pattern(self) -> Optional[str]:
+        """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
+    def resident_children(self) -> tuple:
+        """Process names this CLI keeps as children while idle (#30). The stall
+        watcher does not count them as background work."""
+        return ()
+
+    @property
+    def agent_status_reliable(self) -> bool:
+        """Whether herdr's agent_status for this CLI can drive the idle judgement
+        (#30). On herdr 0.9.1 claude, agy, grok, opencode and codex all report
+        working/idle transitions; a False here turns the idle nudge off."""
+        return True
+
     @property
     def modal_tokens(self) -> Optional[str]:
         return None

.agents/skills/lib_py/agents/input_region.py

@@ -5,36 +5,82 @@ from lib_py.agents.registry import get_adapter
 
 def extract_input_region(pane_text: str, agent_name: str) -> str:
     """
-    Extracts the input area text between horizontal rule separators (─{10,}) for a given agent.
+    Extracts the input area text of a pane, using only the agent adapter's facts
-    If rules are absent, falls back to full pane_text to support mocks and unruled panes.
+    (issue #30, measured on herdr 0.9.1 captures in tests/fixtures/panes):
-    Strips leading prompt tokens ('>', '❯') and placeholders ('Ask anything...').
+
-    Raises ValueError for unsupported agents (like hermes with no input area facts).
+      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` -> from the last line starting (column 0) 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` are stripped, and the prompt and the placeholder
+    (whole-line) are stripped from the box's first non-blank line only. There
+    is no whole-pane fallback: a pane with none of these shapes has no input
+    region and yields a non-empty "unknown" result rather than a guess.
+    Raises ValueError for agents without input area facts (like hermes' peers).
     """
     adapter = get_adapter(agent_name)
     if not adapter or not adapter.input_prompt:
         raise ValueError(f"Agent {agent_name!r} has no input area facts")
 
-    rule_pattern = adapter.input_rule_pattern or r'─{10,}'
-    rule_re = re.compile(rule_pattern)
     lines = pane_text.splitlines()
-    rule_indices = [i for i, line in enumerate(lines) if rule_re.search(line)]
-
-    if len(rule_indices) >= 2:
-        # Exclude the rule lines themselves: lines between rule_indices[-2] + 1 and rule_indices[-1]
-        region_lines = lines[rule_indices[-2] + 1:rule_indices[-1]]
-    elif len(rule_indices) == 1:
-        region_lines = lines[rule_indices[0] + 1:]
-    else:
-        region_lines = lines
-
-    region_text = "\n".join(region_lines)
-
     prompt = adapter.input_prompt
-    if prompt:
+    region_lines = None
-        region_text = re.sub(r'^\s*' + re.escape(prompt) + r'\s*', '', region_text, flags=re.MULTILINE)
 
+    if adapter.input_block_prefix:
+        prefix = adapter.input_block_prefix
+        block = []
+        for line in reversed(lines):
+            if line.lstrip().startswith(prefix):
+                block.insert(0, line.lstrip()[len(prefix):])
+            elif block:
+                break
+        stop_re = re.compile(adapter.input_stop_pattern) if adapter.input_stop_pattern else None
+        for i, line in enumerate(block):
+            if stop_re and stop_re.search(line):
+                block = block[:i]
+                break
+        # No block on screen is not an empty box: fall through to the other shapes.
+        region_lines = block or None
+
+    if region_lines is None and adapter.input_rule_pattern:
+        rule_re = re.compile(adapter.input_rule_pattern)
+        rule_indices = [i for i, line in enumerate(lines) if rule_re.search(line)]
+        if len(rule_indices) >= 2:
+            region_lines = lines[rule_indices[-2] + 1:rule_indices[-1]]
+
+    if region_lines is None:
+        # The composer's prompt sits at column 0; a draft's continuation lines
+        # are indented, so an indented prompt character is draft text (review
+        # ad3d8206), never the start of the input box.
+        starts = [i for i, line in enumerate(lines) if line.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:
+        region_lines = [line.strip().strip(frame) for line in region_lines]
+
+    # The prompt and the placeholder belong to the box's first line only: the
+    # same characters further down are what the user typed.
+    region_lines = list(region_lines)
+    first = next((i for i, line in enumerate(region_lines) if line.strip()), None)
+    if first is not None:
+        head = re.sub(r'^\s*' + re.escape(prompt) + r'\s*', '', region_lines[first], count=1)
         placeholder = adapter.input_placeholder
-    if placeholder:
+        if placeholder and head.strip() == placeholder:
-        region_text = region_text.replace(placeholder, '')
+            head = ''
+        region_lines[first] = head
 
-    return region_text.strip()
+    return "\n".join(region_lines).strip()

.agents/skills/lib_py/delegation/engine.py

@@ -47,7 +47,8 @@ REQUEST_TO_STATUS = {
     "completed": "completed",
     "error": "error",
     "cancelled": "cancelled",
-    "deadline_timeout": "error",
+    "deadline_timeout": "error",      # no event for K x N (subscriber rc 2)
+    "wall_clock_timeout": "error",    # the job budget ran out (subscriber rc 4, Issue #33)
     "merge": None,
     # Operator-authority requests. The wrapper drives multi-iteration delegation
     # (`submit --type loop|discuss`) by re-dispatching a finished job to the next

.agents/skills/lib_py/delegation/job_subscriber.py

@@ -19,7 +19,9 @@ stdout = event lines only. Diagnostics go to stderr via logging.
 Exit codes:
   0  all watched jobs reached ``completed``
   1  any watched job reached ``error``
-  2  timed out (wall-clock or idle) before all jobs finished
+  2  no event for the idle window (K x N) - the wrapper labels it deadline_timeout
+  3  broker unreachable
+  4  the wall-clock budget (--timeout) ran out - the wrapper labels it wall_clock_timeout
 """
 from __future__ import annotations
 
@@ -47,6 +49,16 @@ logger = logging.getLogger("delegate_job.job_subscriber")
 TERMINAL_EVENTS = ("completed", "error")
 TERMINAL_STATUSES = ("completed", "error", "cancelled")
 STATE_EVENT = "state"
+# Exit codes. The wrapper mirrors them as SUB_RC_* (a test keeps the two equal).
+RC_COMPLETED = 0
+RC_ERROR = 1
+# Issue #33: which clock ended the wait. The wrapper maps both to a timeout
+# (caller rc 2) and labels the close from them: rc 2 -> deadline_timeout
+# (no event for the idle window, K x N), rc 4 -> wall_clock_timeout (the job
+# budget, --timeout, ran out while events may still have been arriving).
+RC_NO_EVENT = 2
+RC_BROKER_UNAVAILABLE = 3
+RC_WALL_CLOCK = 4
 
 
 def _format_line(topic: str, payload: Dict[str, Any]) -> str:
@@ -342,9 +354,9 @@ def _run_subscriber(args) -> int:
         jobs = _collect_jobs(args)
     except FileNotFoundError as exc:
         logger.error("%s", exc)
-        return 2
+        return RC_NO_EVENT  # nothing to watch: reported like silence, as before #33
     if not jobs:
-        return 2
+        return RC_NO_EVENT
 
     expected_ids: Set[str] = {j["job_id"] for j in jobs}
     tokens = {j["job_id"]: j.get("auth_token") for j in jobs}
@@ -443,8 +455,8 @@ def _run_subscriber(args) -> int:
     _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts)
     if not pending:
         if any(state == "error" for state in terminal.values()):
-            return 1
+            return RC_ERROR
-        return 0
+        return RC_COMPLETED
 
     try:
         mqtt_common.with_retry(
@@ -459,10 +471,10 @@ def _run_subscriber(args) -> int:
         _check_disk_fallback(pending, args.registry_dir, terminal, dispatch_tokens, attempts)
         if not pending:
             if any(state == "error" for state in terminal.values()):
-                return 1
+                return RC_ERROR
-            return 0
+            return RC_COMPLETED
         logger.error("broker connection failed: %s", exc)
-        return 3
+        return RC_BROKER_UNAVAILABLE
 
     try:
         while pending:
@@ -473,7 +485,7 @@ def _run_subscriber(args) -> int:
                     break
                 logger.error("wall-clock timeout (%.0fs); still pending: %s",
                              wall_timeout, ", ".join(sorted(pending)))
-                return 2
+                return RC_WALL_CLOCK
             if not subscribed:
                 last_event = now
             idle_left = (idle_timeout if first_event_seen else idle_timeout + ack_timeout) - (now - last_event)
@@ -483,7 +495,7 @@ def _run_subscriber(args) -> int:
                     break
                 logger.error("idle timeout (%.0fs, no events); still pending: %s",
                              idle_timeout, ", ".join(sorted(pending)))
-                return 2
+                return RC_NO_EVENT
             wait = min(wall_deadline - now, idle_left, 1.0)
             try:
                 topic, payload = watcher.events.get(timeout=wait)
@@ -544,8 +556,8 @@ def _run_subscriber(args) -> int:
 
     # All jobs reached a terminal state. error wins over completed.
     if any(state == "error" for state in terminal.values()):
-        return 1
+        return RC_ERROR
-    return 0
+    return RC_COMPLETED
 
 
 def main(argv=None) -> int:
@@ -582,7 +594,7 @@ def main(argv=None) -> int:
         return _run_subscriber(args)
     except Exception as exc:
         logger.error("subscriber fatal error: %s", exc)
-        return 3
+        return RC_BROKER_UNAVAILABLE
 
 
 if __name__ == "__main__":

.agents/skills/lib_py/delegation/registry.py

@@ -419,9 +419,9 @@ def _build_parser() -> argparse.ArgumentParser:
     p_update.add_argument("--attempt-id", type=int, default=None)
     p_update.add_argument("--role", default=None)
 
-    p_close = sub.add_parser("close", help="close a job with deadline_timeout or cancelled")
+    p_close = sub.add_parser("close", help="close a job with deadline_timeout, wall_clock_timeout or cancelled")
     p_close.add_argument("--job", required=True)
-    p_close.add_argument("--request", required=True, choices=["deadline_timeout", "cancelled"])
+    p_close.add_argument("--request", required=True, choices=["deadline_timeout", "wall_clock_timeout", "cancelled"])
     p_close.add_argument("--detail", default=None)
     p_close.add_argument("--expect-dispatch-token", default=None)
 

.agents/skills/lib_py/delegation/remote.py

@@ -331,7 +331,7 @@ def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str,
         timeout = float(job.get("timeout_sec", 3600))
         rc, out, err, timed_out = _run_agent_command(exec_cmd, env, timeout)
         if timed_out:
-            detail = "agent command exceeded timeout_sec {}".format(int(timeout))
+            detail = "wall_clock_timeout: agent command exceeded timeout_sec {}".format(int(timeout))
         else:
             tail = (out or "").strip().splitlines()
             detail = tail[-1][:200] if tail else "exit {}".format(rc)
@@ -349,7 +349,7 @@ def handle_dispatch(d: Dict[str, Any], registry_dir: str, session: str,
         if cur.get("status") in ("completed", "error", "cancelled"):
             return 0 if cur.get("status") == "completed" else 1
         time.sleep(1.0)
-    _publish(jid, registry_dir, "error", "deadline_timeout (remote runner)", d)
+    _publish(jid, registry_dir, "error", "wall_clock_timeout (remote runner)", d)
     return 2
 
 

.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
@@ -33,7 +34,7 @@ UNAVAILABLE_MARKER = "!!! CHANGE SET UNAVAILABLE !!!"  # diff_collect.sh MAM_DIF
 # not interrupted and may still be writing; a job that never reached its agent
 # was cancelled before injection.
 TERMINAL = ("completed", "error", "cancelled")
-ORCHESTRATOR_CLOSED = ("deadline_timeout", "agent_blocked")
+ORCHESTRATOR_CLOSED = ("deadline_timeout", "wall_clock_timeout", "agent_blocked")
 SUBJOB_RULE = ("규칙: 열린(OPEN)·미확인 종결(CLOSED-UNCONFIRMED)·실패 하위 잡이 하나라도 있으면 산출물을 미완으로 보고 "
                "[VERDICT: NOT PASS] 를 내십시오. creator 리포트가 해당 부분의 대체 완료를 입증하면 예외입니다. "
                "CLOSED-UNCONFIRMED 는 오케스트레이터가 마감으로 닫은 잡이며 그 에이전트는 계속 일하고 있을 수 있습니다. "
@@ -106,6 +107,7 @@ class LoopRunner:
         self.round_roots: List[str] = []
         self.wait_open_jobs = False
         self.failed_submit_jobs: Set[str] = set()
+        self._stall_seen: Dict[str, Tuple[int, Set[str]]] = {}
         self.current_plan = ""
         self.plan_job_id: Optional[str] = None
         self.base_commit = ""
@@ -121,6 +123,7 @@ class LoopRunner:
         self.reviewer_list: Optional[str] = None
         self.all_reviewers: bool = False
         self.max_loop: int = 3
+        self.job_timeout: Optional[int] = None
         self.max_rebut: int = 1
         self.verbose: bool = False
         self.cleanup: bool = False
@@ -139,6 +142,7 @@ class LoopRunner:
             "  --all-reviewer         Enforce PASS verdict from all active reviewer sessions",
             "  --max-loop N           Max execution-review corrective loop runs (default: 3)",
             "  --max-rebut N          Max rebuttal attempts per review iteration (default: 1, 0: disabled)",
+            "  --job-timeout <sec>    Budget of each delegated job (default: $MAM_JOB_TIMEOUT_S or 3600)",
             "  --verbose              Print detailed execution timeline traces",
             "  --cleanup              Purge temporary job directories upon success",
             "  --worktree <path>      Source worktree for diff/base/head (control plane stays here)",
@@ -209,6 +213,12 @@ class LoopRunner:
                     sys.exit(1)
                 self.max_loop = int(args[i + 1])
                 i += 2
+            elif arg == "--job-timeout":
+                if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]) or int(args[i + 1]) <= 0:
+                    print("ERROR: --job-timeout requires a positive non-zero integer.", file=sys.stderr)
+                    sys.exit(1)
+                self.job_timeout = int(args[i + 1])
+                i += 2
             elif arg == "--max-rebut":
                 if i + 1 >= len(args) or not re.match(r"^[0-9]+$", args[i + 1]):
                     print("ERROR: --max-rebut requires a non-negative integer.", file=sys.stderr)
@@ -454,8 +464,10 @@ class LoopRunner:
     def contract_args(self) -> List[str]:
         """Liveness-contract knobs passed through to every delegate job
         (Issue #18): MAM_HEARTBEAT_SEC (N), MAM_HEARTBEAT_MISSES (K),
-        MAM_ACK_TIMEOUT_S (T). The wrapper derives --idle-timeout as K x N."""
+        MAM_ACK_TIMEOUT_S (T). The wrapper derives --idle-timeout as K x N.
-        args: List[str] = []
+        Issue #33: the job budget is always passed explicitly, so the budget
+        the wrapper registers is the one job_wait_s was derived from."""
+        args: List[str] = ["--timeout", str(self.job_budget)]
         for env_name, flag in (("MAM_HEARTBEAT_SEC", "--heartbeat"),
                                ("MAM_HEARTBEAT_MISSES", "--heartbeat-misses"),
                                ("MAM_ACK_TIMEOUT_S", "--ack-timeout")):
@@ -523,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:
@@ -532,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:
@@ -572,8 +591,27 @@ class LoopRunner:
             return job_id
         return None
 
-    def wait_for_job(self, job_id: str, max_wait: int = 3900, check_interval: float = 1.0) -> bool:
+    @property
+    def job_budget(self) -> int:
+        """Issue #33: --job-timeout > $MAM_JOB_TIMEOUT_S (positive integer) > 3600."""
+        if self.job_timeout:
+            return self.job_timeout
+        env = os.environ.get("MAM_JOB_TIMEOUT_S", "")
+        return int(env) if env.isdigit() and int(env) > 0 else 3600
+
+    @property
+    def job_wait_s(self) -> int:
+        """Wait cap for a job and its sub-jobs: budget + MAM_WAIT_GRACE_S (300,
+        today's 3900 - 3600). The grace covers the wrapper's close/publish tail
+        after the wall clock and the 1 s record poll; it is not proportional
+        to the budget because nothing observed would justify that."""
+        grace = os.environ.get("MAM_WAIT_GRACE_S", "")
+        return self.job_budget + (int(grace) if grace.isdigit() else 300)
+
+    def wait_for_job(self, job_id: str, max_wait: Optional[float] = None, check_interval: float = 1.0) -> bool:
         """Blocking wait for job completion or error with safety timeout."""
+        if max_wait is None:
+            max_wait = self.job_wait_s
         if job_id in self.failed_submit_jobs:
             # Submit command for this job failed (rc!=0). Cap wait to 5s to avoid 3900s hang
             max_wait = min(max_wait, 5.0)
@@ -608,6 +646,7 @@ class LoopRunner:
                     log_error(f"Job '{job_id}' record unreadable / status unknown; aborting wait.")
                     return False
 
+            self._relay_stall_warnings(job_id)
             if status == "completed":
                 if self.verbose:
                     log_success(f"Job '{job_id}' completed successfully.")
@@ -621,6 +660,49 @@ 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.
+        Reads events.log incrementally; never closes the job."""
+        offset, seen = self._stall_seen.get(job_id, (0, set()))
+        try:
+            with open(self.jobs_dir / f"{job_id}.events.log", "r", encoding="utf-8") as fh:
+                fh.seek(offset)
+                lines = fh.readlines()
+                offset = fh.tell()
+        except OSError:
+            return
+        for line in lines:
+            try:
+                ev = json.loads(line)
+            except ValueError:
+                continue
+            detail = str(ev.get("detail") or "")
+            if ev.get("event") == "progress" and detail.startswith(("stalled:", "stall_watch:")) \
+                    and detail not in seen:
+                seen.add(detail)
+                log_warn(f"job {job_id}: {detail}")
+        self._stall_seen[job_id] = (offset, seen)
+
     # ------------------------------------------------------------------
     # Issue #34: a creator job's sub-jobs (registered with --parent-job)
     # ------------------------------------------------------------------
@@ -644,7 +726,10 @@ class LoopRunner:
         A terminal record is not a stopped agent (a deadline close does not
         interrupt it); that is left to the sub-job table and the stale-review
         guard, since nothing signals such an agent's end."""
-        max_wait = float(os.environ.get("MAM_CHILD_WAIT_S", "3900"))
+        child_env = os.environ.get("MAM_CHILD_WAIT_S")
+        max_wait = float(child_env) if child_env else float(self.job_wait_s)
+        wait_src = "MAM_CHILD_WAIT_S" if child_env else \
+            f"MAM_CHILD_WAIT_S default: job budget {self.job_budget} + {self.job_wait_s - self.job_budget}"
         tree_max = int(os.environ.get("MAM_JOB_TREE_MAX", "50"))
         since = self._created_at(root)
         start = time.time()
@@ -667,7 +752,7 @@ class LoopRunner:
                          + " ".join(f"{r.get('job_id')}({r.get('agent_session', '?')}, {r.get('status')})" for r in open_jobs))
                 last = now
             if time.time() - start >= max_wait:
-                log_warn(f"sub-job wait for {root} ended after {int(max_wait)}s (MAM_CHILD_WAIT_S) with "
+                log_warn(f"sub-job wait for {root} ended after {int(max_wait)}s ({wait_src}) with "
                          f"{len(open_jobs)} open; reviewers see them as OPEN")
                 return False
             time.sleep(interval)
@@ -693,7 +778,7 @@ class LoopRunner:
             return "completed", reason or "-"
         # merge review R2 (codex): the remote runner phrases its deadline as
         # "deadline_timeout (remote runner)", so the kind is the first word
-        closed_kind = reason.split(" ", 1)[0] if reason else ""
+        closed_kind = reason.split(" ", 1)[0].rstrip(":") if reason else ""
         if closed_kind in ORCHESTRATOR_CLOSED or (status == "error" and not reason):
             # no recorded reason is not the agent's own report either
             return "CLOSED-UNCONFIRMED", reason or "-"

.agents/skills/lib_py/loop/watcher.py

@@ -74,21 +74,39 @@ def caller_server(environ=None):
     return ""
 
 
-def scoped_env(session=None, environ=None):
+def _names_agent(argv) -> bool:
-    """The environment for a wait aimed at `session`.
+    """True for a herdr call that looks an agent up by name (lib.sh
+    _herdr_scope_is_pane_inherited): `agent ...`, has-session, kill-session."""
+    args = list(argv or [])
+    if args and os.path.basename(args[0]) == "herdr":
+        args = args[1:]
+    while len(args) >= 2 and args[0] in ("-L", "--session"):
+        args = args[2:]
+    return bool(args) and args[0] in ("agent", "has-session", "kill-session")
 
-    Unchanged unless this process runs inside a pane of another server: only
+
-    then are the caller's own workspace/pane/tab ids meaningless for the call
+def scoped_env(session=None, environ=None, argv=None):
-    and dropped. No pane and same-server keep every value.
+    """The environment for a herdr call aimed at `session`.
+
+    Inside a pane of another server the caller's own workspace/pane/tab ids are
+    meaningless for the call and dropped (#23). On the same server, a call that
+    looks an agent up by name (`argv`) drops a HERDR_WORKSPACE_ID that is just
+    the caller pane's own workspace (#34, as mam_herdr does): agent names are
+    unique per server, and a worktree pane must reach an agent of mam-agent.
+    An id that disagrees with the pane, or one set outside a pane, still scopes.
     """
     src = os.environ if environ is None else environ
     caller = caller_server(src)
     target = (session or "").strip()
     # `default` is the same sentinel build_argv reads: no --session flag, so
     # herdr answers on the ambient server - the caller's own.
-    if not caller or not target or target == "default" or caller == target:
+    if caller and target and target != "default" and caller != target:
-        return dict(src)
         return dict((k, v) for k, v in src.items() if k not in CALLER_SCOPE_VARS)
+    ws = src.get("HERDR_WORKSPACE_ID") or ""
+    pane = src.get("HERDR_PANE_ID") or ""
+    if ws and pane and pane.split(":", 1)[0] == ws and _names_agent(argv):
+        return dict((k, v) for k, v in src.items() if k != "HERDR_WORKSPACE_ID")
+    return dict(src)
 
 
 class WaitOutcome(object):
@@ -144,7 +162,7 @@ def wait_once(target: str, until: Sequence[str] = DEFAULT_UNTIL,
     """One barrier call. `run` is injectable so tests never touch a real herdr."""
     argv = build_argv(target, until, timeout_ms, session=session)
     try:
-        proc = run(argv, capture_output=True, text=True, env=scoped_env(session))
+        proc = run(argv, capture_output=True, text=True, env=scoped_env(session, argv=argv))
     except Exception as exc:
         return WaitOutcome(TRANSPORT_ERROR, detail="herdr could not be executed: {}".format(exc))
 

.agents/skills/lib_py/stall_watch.py

@@ -0,0 +1,375 @@
+"""Issue #30: a stall watcher that only notifies.
+
+Runs next to the delegate wrapper's subscriber for a local herdr job and tells
+people when the agent looks stuck. It never closes the job: its only output is
+a `progress` event (through `_notify`, the single publisher) plus log lines,
+and one short nudge prompt per job. Closing stays with the protocol events and
+the two clocks of #18/#33.
+
+  * blocked  -> `progress stalled:blocked <dialog line>` once per episode, no reply.
+  * idle for MAM_STALL_IDLE_S (600) with no background children of the agent
+    process and an empty input box -> `progress stalled:idle` and one nudge.
+  * herdr failing (TRANSPORT_ERROR / AGENT_NOT_FOUND) is never read as "no
+    event": warning + `progress stall_watch:herdr_unavailable:<code>`, backoff,
+    and exit 2 after MAM_STALL_HERDR_FAIL_S (120) without recovery.
+
+Signals are collected in one place (`HerdrSource.collect`) and judged by the
+pure `judge`, so a future remote gateway can hand over the same `Signals`.
+
+Exit codes: 0 the job reached a terminal status, 2 herdr stayed unavailable.
+
+It lives at the top of lib_py, not in `delegation` or `loop`: it needs the
+agent adapters, which neither package may import (their sibling-import rules),
+and it runs as its own process, never inside a frozen loop.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import logging
+import os
+import re
+import signal
+import subprocess
+import sys
+import time
+from dataclasses import dataclass, field
+from typing import List, Optional
+
+from lib_py.agents.input_region import extract_input_region
+from lib_py.agents.registry import agent_of_row, get_adapter
+from lib_py.delegation.engine import TERMINAL_STATUSES
+from lib_py.loop import watcher
+
+log = logging.getLogger("stall_watch")
+
+RC_TERMINAL = 0
+RC_HERDR_UNAVAILABLE = 2
+
+NOTIFY_BLOCKED = "notify_blocked"
+NOTIFY_IDLE_AND_NUDGE = "notify_idle_and_nudge"
+IDLE_STATES = ("idle", "done")
+HEARTBEAT_MARK = "lib_py.delegation.heartbeat"
+SKILLS_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+NUDGE = ("Job {job}: you look idle with no background work. If the job is done, publish "
+         "completed or error with the command in your brief; otherwise continue where you left off.")
+
+
+@dataclass
+class Signals:
+    agent_status: Optional[str] = None
+    state_change_seq: Optional[int] = None
+    pane_id: Optional[str] = None
+    blocked_text: str = ""
+    agent_pid: Optional[int] = None
+    background_children: List[str] = field(default_factory=list)
+    input_empty: Optional[bool] = None
+    job_status: Optional[str] = None
+    idle_for: float = 0.0
+
+
+@dataclass
+class WatchState:
+    idle_s: float = 600.0
+    reliable: bool = True
+    nudged: bool = False
+    blocked_notified: bool = False
+
+
+def judge(sig: Signals, st: WatchState) -> Optional[str]:
+    """Pure decision over one set of signals. Never an action that closes a job."""
+    if sig.job_status in TERMINAL_STATUSES:
+        return None
+    if sig.agent_status == "blocked":
+        return None if st.blocked_notified else NOTIFY_BLOCKED
+    if (st.reliable and not st.nudged and sig.agent_status in IDLE_STATES
+            and sig.idle_for >= st.idle_s and sig.agent_pid
+            and not sig.background_children and sig.input_empty is True):
+        return NOTIFY_IDLE_AND_NUDGE
+    return None
+
+
+def _proc_table():
+    """pid -> (ppid, name, cmdline) from /proc (Linux); `ps` elsewhere."""
+    table = {}
+    if os.path.isdir("/proc/self"):
+        for d in os.listdir("/proc"):
+            if not d.isdigit():
+                continue
+            try:
+                with open("/proc/%s/stat" % d) as f:
+                    stat = f.read()
+                with open("/proc/%s/cmdline" % d, "rb") as f:
+                    argv = [a.decode("utf-8", "replace") for a in f.read().split(b"\0") if a]
+            except OSError:
+                continue
+            name = stat[stat.index("(") + 1:stat.rindex(")")]
+            ppid = int(stat[stat.rindex(")") + 2:].split()[1])
+            table[int(d)] = (ppid, name, argv)
+        return table
+    out = subprocess.run(["ps", "-eo", "pid=,ppid=,comm=,args="], capture_output=True, text=True).stdout
+    for line in out.splitlines():
+        parts = line.split(None, 3)
+        if len(parts) >= 3:
+            table[int(parts[0])] = (int(parts[1]), os.path.basename(parts[2]), (parts[3] if len(parts) > 3 else parts[2]).split())
+    return table
+
+
+def background_children(agent_pid: int, resident=()) -> List[str]:
+    """Descendants of the agent process, minus the adapter's resident children
+    and the heartbeat timer (decision 9: it must never count as work)."""
+    table = _proc_table()
+    kids = {}
+    for pid, (ppid, _, _) in table.items():
+        kids.setdefault(ppid, []).append(pid)
+    found, todo = [], list(kids.get(agent_pid, []))
+    while todo:
+        pid = todo.pop()
+        todo.extend(kids.get(pid, []))
+        _, name, argv = table[pid]
+        base = os.path.basename(argv[0]) if argv else ""
+        if name in resident or base in resident or name[:15] in [r[:15] for r in resident]:
+            continue
+        if HEARTBEAT_MARK in " ".join(argv):
+            continue
+        found.append("%d:%s" % (pid, name))
+    return found
+
+
+def resolve_agent(agent, target):
+    """Adapter name for the wrapper's --agent (its default is `claude-code`),
+    else from the session name; None when no adapter matches."""
+    for name in (agent, (agent or "").split("-")[0], agent_of_row({}, session_name=target)):
+        if name and get_adapter(name):
+            return get_adapter(name).name
+    return None
+
+
+class HerdrError(Exception):
+    def __init__(self, code):
+        super().__init__(code)
+        self.code = code
+
+
+class HerdrSource(object):
+    """Every herdr and OS read the judgement needs, in one place."""
+
+    def __init__(self, target, agent, session=None):
+        self.target, self.session = target, session or None
+        self.agent = resolve_agent(agent, target)
+        adapter = get_adapter(self.agent) if self.agent else None
+        self.resident = tuple(adapter.resident_children) if adapter else ()
+
+    def _herdr(self, *args):
+        argv = ["herdr"] + (["--session", self.session] if self.session and self.session != "default" else []) + list(args)
+        try:
+            proc = subprocess.run(argv, capture_output=True, text=True, timeout=30,
+                                  env=watcher.scoped_env(self.session, argv=argv))
+        except Exception as exc:
+            raise HerdrError("exec_failed: %s" % exc)
+        if proc.returncode != 0:
+            try:
+                code = json.loads(proc.stderr)["error"]["code"]
+            except Exception:
+                code = "rc%d" % proc.returncode
+            raise HerdrError(code)
+        return proc.stdout
+
+    def agent_get(self):
+        payload = json.loads(self._herdr("agent", "get", self.target) or "{}")
+        return (payload.get("result") or {}).get("agent") or {}
+
+    def agent_pid(self, pane_id):
+        info = json.loads(self._herdr("pane", "process-info", "--pane", pane_id) or "{}")
+        procs = ((info.get("result") or {}).get("process_info") or {}).get("foreground_processes") or []
+        # Not shell_pid: the shell is not the agent; without a pid we withhold.
+        return procs[0].get("pid") if procs else None
+
+    def blocked_text(self, pane_id):
+        text = self._herdr("pane", "read", pane_id, "--source", "detection", "--lines", "8")
+        for line in text.splitlines():
+            line = re.sub(r"[^\x20-\x7e]+", " ", line).strip()
+            line = re.sub(r"\s+", " ", line)
+            if line:
+                return line[:160]
+        return ""
+
+    def input_empty(self):
+        """True/False from the adapter's input-box facts, None when it has none."""
+        if not self.agent:
+            return None
+        try:
+            return extract_input_region(self._herdr("agent", "read", self.target), self.agent) == ""
+        except ValueError:
+            return None
+
+    def prompt(self, text):
+        self._herdr("agent", "prompt", self.target, text)
+
+    def collect(self, sig: Signals) -> Signals:
+        """Fill the process and input signals of an idle agent."""
+        agent = self.agent_get()
+        sig.agent_status = agent.get("agent_status", sig.agent_status)
+        sig.state_change_seq = agent.get("state_change_seq", sig.state_change_seq)
+        sig.pane_id = agent.get("pane_id") or sig.pane_id
+        sig.agent_pid = self.agent_pid(sig.pane_id) if sig.pane_id else None
+        sig.background_children = background_children(sig.agent_pid, self.resident) if sig.agent_pid else []
+        sig.input_empty = self.input_empty()
+        return sig
+
+
+class StallWatch(object):
+    def __init__(self, args, source=None):
+        self.args = args
+        self.src = source or HerdrSource(args.target, args.agent, args.herdr_session)
+        name = resolve_agent(args.agent, args.target)
+        adapter = get_adapter(name) if name else None
+        reliable = bool(adapter.agent_status_reliable) if adapter else True
+        if os.environ.get("MAM_TEST_UNRELIABLE_AGENT") == args.agent:
+            reliable = False
+        self.st = WatchState(idle_s=float(os.environ.get("MAM_STALL_IDLE_S", "600")), reliable=reliable)
+        self.fail_s = float(os.environ.get("MAM_STALL_HERDR_FAIL_S", "120"))
+        self.once = set()
+
+    def _notify(self, detail):
+        """The only publisher: a `progress` event, nothing else."""
+        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:
+            cmd += ["--expect-dispatch-token", self.args.expect_dispatch_token]
+        if self.args.attempt_id is not None:
+            cmd += ["--attempt-id", str(self.args.attempt_id)]
+        env = dict(os.environ)
+        env["PYTHONPATH"] = SKILLS_DIR + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "")
+        # rc 2 = broker unreachable after the local record and events.log were written.
+        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:])
+
+    def _notify_once(self, key, detail, message):
+        if key in self.once:
+            return
+        self.once.add(key)
+        log.warning(message)
+        self._notify(detail)
+
+    def _job_status(self):
+        try:
+            with open(os.path.join(self.args.registry_dir, self.args.job + ".json")) as f:
+                return json.load(f).get("status")
+        except (OSError, ValueError):
+            return None
+
+    def run(self) -> int:
+        st, sig, agent = self.st, Signals(), self.args.agent
+        if not st.reliable:
+            log.warning("stall_watch: idle judgement off for %s (agent_status unreliable)", agent)
+        backoff = watcher.ReconnectBackoff()
+        fail_since, idle_since = None, None
+        while True:
+            sig.job_status = self._job_status()
+            if sig.job_status in TERMINAL_STATUSES:
+                return RC_TERMINAL
+            if sig.agent_status is None:
+                until, tmo = watcher.VALID_STATES, 1000  # learn the current state
+            elif sig.agent_status in IDLE_STATES:
+                left = st.idle_s - (time.monotonic() - idle_since)
+                # Past S the judgement repeats every 30 s, not on every tick.
+                until, tmo = ("working", "blocked"), int(1000 * (min(left, 30.0) if left > 0 else 30.0))
+                tmo = max(tmo, 1000)
+            elif sig.agent_status == "blocked":
+                until, tmo = ("idle", "done", "working"), 60000
+            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 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()
+                if fail_since is None:
+                    fail_since = now
+                    log.warning("stall_watch: herdr unavailable: %s %s", code, out.detail)
+                    self._notify("stall_watch:agent_not_found" if code == "agent_not_found"
+                                 else "stall_watch:herdr_unavailable:%s" % code)
+                if now - fail_since >= self.fail_s:
+                    log.warning("stall_watch: herdr unavailable for %ds; stall watch stops (%s)", int(now - fail_since), code)
+                    return RC_HERDR_UNAVAILABLE
+                time.sleep(min(backoff.record_failure(), max(0.1, self.fail_s - (now - fail_since))))
+                sig.agent_status = None  # re-observe after a gap
+                continue
+            backoff.record_success()
+            fail_since = None
+            if out.kind == watcher.MATCHED:
+                new = out.agent.get("agent_status")
+                sig.pane_id = out.agent.get("pane_id") or sig.pane_id
+                if new in IDLE_STATES and sig.agent_status not in IDLE_STATES:
+                    idle_since, sig.state_change_seq = time.monotonic(), out.agent.get("state_change_seq")
+                if new != "blocked":
+                    st.blocked_notified = False
+                sig.agent_status = new
+            try:
+                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", ""))
+            if judge(sig, st) == NOTIFY_BLOCKED:
+                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 idle_since
+        if sig.agent_status not in IDLE_STATES or not st.reliable or st.nudged:
+            return idle_since
+        sig.idle_for = time.monotonic() - idle_since
+        if sig.idle_for < st.idle_s:
+            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:
+            # 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:
+            self._notify_once("input_unavailable", "stall_watch:input_check_unavailable:%s" % agent,
+                              "stall_watch: input check unavailable for %s; idle nudge withheld" % agent)
+        elif sig.input_empty is False:
+            log.info("stall_watch: input box of %s is not empty; nudge withheld", self.args.target)
+            if sig.idle_for >= 3 * st.idle_s:
+                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:
+            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
+            self.src.prompt(NUDGE.format(job=self.args.job))
+        return idle_since
+
+
+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)
+    p.add_argument("--registry-dir", required=True)
+    p.add_argument("--target", required=True, help="herdr agent name of the job's session")
+    p.add_argument("--agent", required=True, help="agent CLI (adapter name)")
+    p.add_argument("--herdr-session", default="")
+    p.add_argument("--expect-dispatch-token", default=None)
+    p.add_argument("--attempt-id", type=int, default=None)
+    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.
+    signal.signal(signal.SIGTERM, lambda *_: os._exit(RC_TERMINAL))
+    return StallWatch(args).run()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

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

@@ -46,6 +46,8 @@ multi-agent-mux-delegate-job submit \
   #   must arrive within T s of the injection or the inject fails closed.
   #   --prompt-file <path> replaces --prompt for long prompts (Issue #36): one
   #   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).
   #   --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
@@ -95,6 +97,7 @@ 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. |
 | `5` | Session Busy | `cancelled` (`session_busy`) | Precondition failed / `session_busy`: Preflight check (`preflight_agent`) or TUI idle handshake wait failed (prompt never delivered, retryable). |
 

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

@@ -111,7 +111,7 @@ usage() {
 multi-agent-mux-delegate-job <command> [options]
 
   submit  --agent <name> (--prompt <text> | --prompt-file <path>) [--workdir <dir>] [--agent-session <label>]
-          [--timeout <sec>] [--idle-timeout <sec>] [--validate <script>]
+          [--timeout <sec> (default: $MAM_JOB_TIMEOUT_S or 3600)] [--idle-timeout <sec>] [--validate <script>]
           [--heartbeat <sec>] [--heartbeat-misses <count>] [--ack-timeout <sec>]
           [--registry-dir <dir>] [--dry-run] [--role <role_name>]
           [--type <direct|loop|discuss>] [--reviewer <reviewer_agent>]
@@ -140,6 +140,16 @@ EOF
 # ---- arg parsing helpers --------------------------------------------------
 AGENT="claude-code"; PROMPT=""; PROMPT_FILE=""; WORKDIR="$(pwd)"; AGENT_SESSION="herdr:claude"
 TIMEOUT=3600; IDLE_TIMEOUT=""; VALIDATE=""; DRY_RUN=0
+# job_subscriber.py exit codes (its RC_* constants; a test keeps the two equal).
+SUB_RC_COMPLETED=0; SUB_RC_ERROR=1; SUB_RC_NO_EVENT=2; SUB_RC_BROKER_UNAVAILABLE=3; SUB_RC_WALL_CLOCK=4
+# Issue #33: the job budget falls back to $MAM_JOB_TIMEOUT_S (--timeout > env > 3600).
+if [[ -n "${MAM_JOB_TIMEOUT_S:-}" ]]; then
+  if [[ "$MAM_JOB_TIMEOUT_S" =~ ^[0-9]+$ && "$MAM_JOB_TIMEOUT_S" -gt 0 ]]; then
+    TIMEOUT="$MAM_JOB_TIMEOUT_S"
+  else
+    echo "WARNING: MAM_JOB_TIMEOUT_S=$MAM_JOB_TIMEOUT_S ignored (not a positive integer); using 3600" >&2
+  fi
+fi
 # Issue #18 liveness contract (N / K / T). An unset --idle-timeout derives K x N.
 HEARTBEAT_SEC="${MAM_HEARTBEAT_SEC:-30}"; HEARTBEAT_MISSES="${MAM_HEARTBEAT_MISSES:-4}"; ACK_TIMEOUT="${MAM_ACK_TIMEOUT_S:-180}"
 JOB_ID=""; REGISTRY_DIR="$REGISTRY_DIR_DEFAULT"; DELEGATE_ROLE="Worker"
@@ -281,7 +291,7 @@ EOF
 
     # 1) Preflight (a remote session has no local pane to check: Issue #18)
     if ! _is_remote "$AGENT_SESSION" && ! preflight_agent "$JOB_ID" "$AGENT_SESSION" 0; then
-      finalize_unfinished "$JOB_ID" "error" 1 "preflight"
+      finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight"
       local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
       echo "$logs_root/$JOB_ID"
       exit 5
@@ -326,7 +336,7 @@ EOF
       else
         local sub_exit=0
         wait "$sub_pid" 2>/dev/null || sub_exit=$?
-        if [ $sub_exit -eq 0 ] || [ $sub_exit -eq 3 ]; then
+        if [ $sub_exit -eq $SUB_RC_COMPLETED ] || [ $sub_exit -eq $SUB_RC_BROKER_UNAVAILABLE ]; then
           sub_ready=1
           break
         else
@@ -352,11 +362,12 @@ EOF
     # 5) Deliver the prompt (herdr pane or remote dispatch) and wait for `started`
     if ! deliver_prompt "$JOB_ID" "$instructions" "$AGENT_SESSION" "$logf"; then
       reap_subscriber "$sub_pid"
-      finalize_unfinished "$JOB_ID" "error" 1 "inject"
+      finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
       local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
       echo "$logs_root/$JOB_ID"
       exit 1
     fi
+    start_stall_watch "$AGENT_SESSION" "$AGENT" "$_disp_tok" "$_attempt_id"
 
     # 6) optional validation hook
     if [[ -n "$VALIDATE" ]]; then
@@ -371,6 +382,7 @@ EOF
 
     local sub_rc=0
     wait "$sub_pid" || sub_rc=$?
+    stop_stall_watch
     echo "subscriber output:"; cat "$logf" || true
 
     local job_status
@@ -446,7 +458,7 @@ EOF
 
       # Preflight (skipped for a remote session: Issue #18)
       if ! _is_remote "$current_session" && ! preflight_agent "$JOB_ID" "$current_session" "$force_warn_only"; then
-        finalize_unfinished "$JOB_ID" "error" 1 "preflight"
+        finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "preflight"
         local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
         echo "$logs_root/$JOB_ID"
         exit 5
@@ -502,7 +514,7 @@ EOF
         else
           local sub_exit=0
           wait "$sub_pid" 2>/dev/null || sub_exit=$?
-          if [ $sub_exit -eq 0 ] || [ $sub_exit -eq 3 ]; then
+          if [ $sub_exit -eq $SUB_RC_COMPLETED ] || [ $sub_exit -eq $SUB_RC_BROKER_UNAVAILABLE ]; then
             sub_ready=1
             break
           else
@@ -535,15 +547,17 @@ EOF
       # Trigger agent
       if ! deliver_prompt "$JOB_ID" "$instructions" "$current_session" "$logf"; then
         reap_subscriber "$sub_pid"
-        finalize_unfinished "$JOB_ID" "error" 1 "inject"
+        finalize_unfinished "$JOB_ID" "error" "$SUB_RC_ERROR" "inject"
         local logs_root="${DELEGATE_JOB_LOGS_DIR:-$WORKDIR/delegate_job_logs}"
         echo "$logs_root/$JOB_ID"
         exit 1
       fi
+      start_stall_watch "$current_session" "$cur_agent" "$_disp_tok" "$attempt"
 
       # Wait for subscriber
       local sub_rc=0
       wait "$sub_pid" || sub_rc=$?
+      stop_stall_watch
       echo "subscriber output:"; cat "$logf" || true
 
       local job_status
@@ -664,6 +678,47 @@ reap_subscriber() {
   fi
 }
 
+# Issue #30: a notify-only stall watcher for a job with a local herdr pane
+# (lib_py.stall_watch decides; it never closes the job). A remote
+# or human job has no pane to watch. Sets STALL_PID ("" when none started).
+start_stall_watch() {
+  local session="$1" agent="$2" tok="${3:-}" att="${4:-}"
+  STALL_PID=""
+  [ "${MAM_STALL_WATCH:-1}" = "0" ] && return 0
+  if _is_remote "$session"; then
+    echo "stall watch: remote job, no local pane - stall detection off" >&2; return 0
+  fi
+  if [ "$agent" = "human" ] || [ "${session#herdr:}" = "human" ]; then
+    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:-}")
+  [ -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[@]}" \
+      >"$REGISTRY_DIR/$JOB_ID.stall.out" 2>&1 &
+  STALL_PID=$!
+  echo "stall watch pid: $STALL_PID (log: $REGISTRY_DIR/$JOB_ID.stall.out)"
+}
+
+# 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.
+stop_stall_watch() {
+  [ -n "${STALL_PID:-}" ] || return 0
+  local rc=0
+  if kill -0 "$STALL_PID" 2>/dev/null; then
+    kill "$STALL_PID" 2>/dev/null || true
+    wait "$STALL_PID" 2>/dev/null || true
+  else
+    wait "$STALL_PID" 2>/dev/null || rc=$?
+  fi
+  if [ "$rc" -ne 0 ]; then
+    echo "stall watch for $JOB_ID exited rc=$rc:" >&2
+    tail -n 3 "$REGISTRY_DIR/$JOB_ID.stall.out" >&2 2>/dev/null || true
+  fi
+  STALL_PID=""
+}
+
 _disk_status() {
   local jid="$1"
   "$PY" -c "import json, os; p=os.path.join('$REGISTRY_DIR', '$jid.json'); print(json.load(open(p)).get('status','')) if os.path.exists(p) else print('')" 2>/dev/null || true
@@ -774,11 +829,11 @@ PYEOF
 _map_sub_rc() {
   local sub_rc="$1" job_id="$2"
   local job_status="running"
-  if [[ $sub_rc -eq 0 ]]; then
+  if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then
     job_status="completed"
-  elif [[ $sub_rc -eq 1 ]]; then
+  elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then
     job_status="error"
-  elif [[ $sub_rc -eq 3 ]]; then
+  elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then
     job_status="broker_unavailable"
     local disk_st
     disk_st="$(_disk_status "$job_id")"
@@ -786,7 +841,7 @@ _map_sub_rc() {
       job_status="$disk_st"
     fi
   else
-    job_status="timeout"
+    job_status="timeout"  # SUB_RC_NO_EVENT, SUB_RC_WALL_CLOCK (and any other code)
   fi
   echo "$job_status"
 }
@@ -803,13 +858,18 @@ _rc_for() {
 }
 
 finalize_unfinished() {
-  local job_id="$1" job_status="$2" sub_rc="$3" phase="${4:-}"
+  local job_id="$1" job_status="$2" sub_rc="$3" phase="${4:-}" FINALIZE_CLOCK=""
   if [ "${MAM_DELEGATE_FAIL_CLOSED:-1}" = "0" ]; then
     return 0
   fi
   if [ "$job_status" = "completed" ]; then
     return 0
   fi
+  # Issue #30: what the stall watcher saw goes with an unfinished job's close.
+  if [ -s "$REGISTRY_DIR/$job_id.stall.out" ]; then
+    echo "stall watch log for $job_id (last 3 lines):" >&2
+    tail -n 3 "$REGISTRY_DIR/$job_id.stall.out" >&2 || true
+  fi
 
   local cur_disk_st="$(_disk_status "$job_id")"
   if [ "$cur_disk_st" = "completed" ] || [ "$cur_disk_st" = "error" ] || [ "$cur_disk_st" = "cancelled" ]; then
@@ -838,6 +898,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
@@ -885,9 +947,16 @@ finalize_unfinished() {
       [ "$left" -le 0 ] && break
       sleep 1
     done
+    [ "$(date +%s)" -ge "$deadline" ] && FINALIZE_CLOCK="wall"
   fi
 
+  # Issue #33: the subscriber's rc says which clock ran out (4 wall clock,
+  # 2 no event); the disk-heartbeat grace above that ends on the deadline is
+  # the wall clock too. bash only carries the label.
+  local clock="${FINALIZE_CLOCK:-idle}"
+  [ "$sub_rc" = "$SUB_RC_WALL_CLOCK" ] && clock="wall"
   local close_req="deadline_timeout"
+  [ "$clock" = "wall" ] && [ "$phase" != "inject" ] && close_req="wall_clock_timeout"
   local detail=""
   if [ "$phase" = "preflight" ] || [ "$phase" = "idle_wait" ]; then
     close_req="cancelled"
@@ -896,6 +965,7 @@ finalize_unfinished() {
     [ "$phase" = "preflight" ] && [ -n "${PREFLIGHT_REASON:-}" ] && detail="$PREFLIGHT_REASON"
   fi
 
+  [ "$close_req" != "cancelled" ] && echo "finalize: $job_id closed by $([ "$clock" = "wall" ] && echo wall_clock || echo no_event) clock (budget ${TIMEOUT}s / idle ${IDLE_TIMEOUT}s)" >&2
   local close_args=(--job "$job_id" --request "$close_req")
   [ -n "$detail" ] && close_args+=(--detail "$detail")
   [ -n "${_disp_tok:-}" ] && close_args+=(--expect-dispatch-token "$_disp_tok")
@@ -924,8 +994,8 @@ finalize_unfinished() {
     [ -n "${_disp_tok:-}" ] && disp_arg=(--expect-dispatch-token "$_disp_tok")
     if [ "$close_req" = "cancelled" ]; then
       "$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$job_id" --event cancelled --detail "$detail" ${disp_arg[@]+"${disp_arg[@]}"} 2>/dev/null || true
-    elif [ "$sub_rc" -ne 1 ] || [ "$phase" = "inject" ]; then
+    elif [ "$sub_rc" -ne "$SUB_RC_ERROR" ] || [ "$phase" = "inject" ]; then
-      local err_detail="deadline_timeout"
+      local err_detail="$close_req"
       [ "$phase" = "inject" ] && err_detail="agent_blocked"
       "$PY" "$SCRIPT_DIR/scripts/publish_event.py" --registry-dir "$REGISTRY_DIR" --job "$job_id" --event error --detail "$err_detail" ${disp_arg[@]+"${disp_arg[@]}"} 2>/dev/null || true
     fi
@@ -1131,13 +1201,18 @@ cmd_logs() {
 cmd_wait() {
   parse_opts "$@"
   PY="$(pick_python)"
+  local rc=0
   if [[ -n "$JOB_ID" ]]; then
     "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
-        --job "$JOB_ID" --timeout "$TIMEOUT"
+        --job "$JOB_ID" --timeout "$TIMEOUT" || rc=$?
   else
     "$PY" "$SCRIPT_DIR/scripts/job_subscriber.py" --registry-dir "$REGISTRY_DIR" \
-        --wait-any --timeout "$TIMEOUT"
+        --wait-any --timeout "$TIMEOUT" || rc=$?
   fi
+  # Issue #33: either clock is a timeout (rc 2) to callers of `wait`, as before;
+  # the subscriber's own log says which clock it was.
+  [ "$rc" -eq "$SUB_RC_WALL_CLOCK" ] && rc=$SUB_RC_NO_EVENT
+  return "$rc"
 }
 
 main() {

.agents/skills/multi-agent-mux-loop/SKILL.md

@@ -206,6 +206,7 @@ sequenceDiagram
 | **Phase 3: Consensus** | `--all-reviewer` | 레지스트리에 등록된 모든 active 리뷰어 세션을 자동으로 수집하여 리뷰를 돌립니다. (`--reviewer` 옵션과는 상호 배타적이며, 지정/수집된 모든 리뷰어의 PASS 만장일치가 항상 필요합니다.) |
 | **Iterative Loop** | `--max-loop M` | NOT PASS 판정 시 최대 `M`회까지 Creator가 자체 수정합니다. `--plan` 모드에서 리뷰어가 리포트에 `[ESCALATE: PLANNER]` 태그를 남기면 설계 변경 수준으로 판단하여 Planner에게 계획 갱신을 위임합니다 (린트는 리뷰어가 검토 관점 중 하나로 확인할 뿐, 별도의 자동 게이트는 아닙니다). |
 | **Rebuttal** | `--max-rebut N` | 이터레이션당 Creator 반론 횟수 (기본 1, `0`이면 끔). |
+| **Job budget** | `--job-timeout <sec>` | 위임 잡마다의 예산 (기본 `MAM_JOB_TIMEOUT_S` 또는 3600). 잡 대기 상한은 예산 + 300 s. |
 | **Control** | `--verbose` / `--cleanup` | 상세 로그 / 성공 시 임시 job 디렉터리 삭제. |
 | **Source root** | `--worktree <path>` (env `MAM_LOOP_WORKTREE`, CLI 우선) | diff 수집·base·head·빈 변경분 판정만 이 git 워크트리 기준으로 합니다. wrapper·레지스트리·체크포인트·loop guard·리포트 승격·cleanup 은 루프를 띄운 본 체크아웃(제어 루트)에 남습니다. 모든 위임 프롬프트 앞에 `[워크트리 컨텍스트]` 블록(경로·브랜치·base·head·커밋 수·미커밋 변경)이 붙습니다. Python 루프 전용(Issue #21). |
 | **Diff base** | `--base <ref>` (CLI 전용, env 없음) | 리뷰 diff 의 base 를 고정합니다. 소스 HEAD 의 조상이어야 하며 아니면 exit 1. 생략 시 Phase 2 시작 시점의 소스 HEAD 입니다. |
@@ -218,7 +219,7 @@ sequenceDiagram
 
 **리뷰 identity (Issue #21 항목 5).** `--worktree`/`--base` 가 있으면 매 리뷰 라운드를 `(loop_id, base, head, tree)` 에 묶습니다. 소스의 미커밋·untracked 상태까지 담은 스냅샷(ref 없는 커밋)을 만들고, 리뷰어가 받는 diff 도 그 스냅샷에서 만듭니다(`base..snapshot`). 라운드가 끝난 뒤 소스가 바뀌었으면 그 라운드의 PASS 는 버리고 수정 잡 없이 다시 리뷰합니다(마지막 반복이면 exit 1). 승격 직전에도 한 번 더 확인하며, 승격된 리포트 옆에 `report-<jid>.identity.json`(loop_id·base·head·tree·스냅샷 커밋·리뷰어·`transport`·`evidence`: `review-tree`/`source-tree`/`diff-only`)을 둡니다. 추적되지 않는 중첩 git 저장소는 스냅샷에 담을 수 없어 거부합니다. 서브모듈 안의 미커밋·untracked 내용도 스냅샷에는 gitlink 커밋만 남으므로 거부합니다(라운드 시작 시 exit 1, 리뷰 중에 생기면 그 라운드는 stale). `submodule.*.ignore` 설정으로 가려지지 않습니다. 리뷰어가 없으면 creator 셀프 리뷰가 같은 리뷰어 경로(스냅샷 diff·증거 검사·승격 직전 재확인·사이드카)를 거치고, NOT PASS 면 수정 잡 없이 다시 리뷰합니다. 리뷰 잡의 `[워크트리 컨텍스트]` 는 검증 대상을 `[리뷰 identity]` 의 검증 경로 하나로만 지시합니다.
 
-**하위 위임 잡 (Issue #34).** creator 가 작업 일부를 `delegate-job submit --parent-job <자기 job_id>` 로 맡기면, 루프는 creator 잡(실행·수정·순응)이 끝난 뒤 그 잡의 자손(자식·손자…, 폴링마다 다시 읽음)의 레코드가 모두 종결될 때까지 기다린 다음 리뷰로 넘어갑니다. 깊이 제한은 없고 상한은 둘입니다: `MAM_CHILD_WAIT_S`(기본 3900 s, creator 잡 종결부터; 각 하위 잡은 자기 등록 + `timeout_sec`(#33, 기본 3600 s)에 닫히므로 보통 그 전에 끝납니다)와 `MAM_JOB_TREE_MAX`(기본 50 개, 넘으면 기다리지 않음). 그만큼 루프 벽시계가 늘어납니다. 루프를 잡으로 감싸 돌리면 바깥 잡의 3600 s 가 먼저 끝날 수 있습니다. 자식의 결과는 루프를 멈추지 않고 리뷰 근거가 됩니다: 하위 잡이나 귀속되지 않은 열린 잡이 있으면 모든 리뷰 프롬프트(셀프 리뷰 포함) 앞에 `[하위 위임 잡]` 표(상태·사유·깊이)와 `[귀속되지 않은 열린 잡]`(부모 미표기로 이 작업 중 등록된 열린 잡) 줄이 붙고(둘 다 없으면 프롬프트는 전과 같습니다), 리뷰어는 `OPEN`·`CLOSED-UNCONFIRMED`·실패(`error`/`cancelled`/`NOT-DISPATCHED`) 하위 잡이 있으면 creator 리포트가 대체 완료를 입증하지 않는 한 NOT PASS 를 냅니다. 표는 이 실행의 모든 creator 잡(앞 회차의 실행·수정·순응 잡 포함)의 자손을 매 회차 다시 읽어 만듭니다 — 수정 잡이 새로 생겼다고 앞 잡의 열린·실패 하위 잡이 사라지지 않습니다(대기는 새 creator 잡의 자손만 하므로 대기 상한이 회차마다 다시 쌓이지 않습니다). `CLOSED-UNCONFIRMED` 는 오케스트레이터가 마감(`deadline_timeout`/`agent_blocked`)으로 닫았거나 사유가 없는 `error` 입니다 — 레코드만 닫혔고 에이전트는 인터럽트되지 않으므로 계속 쓰고 있을 수 있습니다. creator 잡 자체의 실패는 전처럼 exit 1 입니다.
+**하위 위임 잡 (Issue #34).** creator 가 작업 일부를 `delegate-job submit --parent-job <자기 job_id>` 로 맡기면, 루프는 creator 잡(실행·수정·순응)이 끝난 뒤 그 잡의 자손(자식·손자…, 폴링마다 다시 읽음)의 레코드가 모두 종결될 때까지 기다린 다음 리뷰로 넘어갑니다. 깊이 제한은 없고 상한은 둘입니다: `MAM_CHILD_WAIT_S`(기본 잡 예산 + 300 s, creator 잡 종결부터; 각 하위 잡은 자기 등록 + `timeout_sec`(#33, 기본 `MAM_JOB_TIMEOUT_S`/3600 s)에 닫히므로 보통 그 전에 끝납니다)와 `MAM_JOB_TREE_MAX`(기본 50 개, 넘으면 기다리지 않음). 그만큼 루프 벽시계가 늘어납니다. 루프를 잡으로 감싸 돌리면 바깥 잡의 3600 s 가 먼저 끝날 수 있습니다. 자식의 결과는 루프를 멈추지 않고 리뷰 근거가 됩니다: 하위 잡이나 귀속되지 않은 열린 잡이 있으면 모든 리뷰 프롬프트(셀프 리뷰 포함) 앞에 `[하위 위임 잡]` 표(상태·사유·깊이)와 `[귀속되지 않은 열린 잡]`(부모 미표기로 이 작업 중 등록된 열린 잡) 줄이 붙고(둘 다 없으면 프롬프트는 전과 같습니다), 리뷰어는 `OPEN`·`CLOSED-UNCONFIRMED`·실패(`error`/`cancelled`/`NOT-DISPATCHED`) 하위 잡이 있으면 creator 리포트가 대체 완료를 입증하지 않는 한 NOT PASS 를 냅니다. 표는 이 실행의 모든 creator 잡(앞 회차의 실행·수정·순응 잡 포함)의 자손을 매 회차 다시 읽어 만듭니다 — 수정 잡이 새로 생겼다고 앞 잡의 열린·실패 하위 잡이 사라지지 않습니다(대기는 새 creator 잡의 자손만 하므로 대기 상한이 회차마다 다시 쌓이지 않습니다). `CLOSED-UNCONFIRMED` 는 오케스트레이터가 마감(`deadline_timeout`/`wall_clock_timeout`/`agent_blocked`)으로 닫았거나 사유가 없는 `error` 입니다 — 레코드만 닫혔고 에이전트는 인터럽트되지 않으므로 계속 쓰고 있을 수 있습니다. creator 잡 자체의 실패는 전처럼 exit 1 입니다.
 
 **리뷰 중 소스 변경 = 승인 무효 (하위 잡이 있을 때, Issue #34).** `--worktree`/`--base` 가 없어도 이 실행에 하위 잡(또는 귀속되지 않은 열린 잡)이 있는 라운드마다 리뷰 요청 직전 소스(= 제어 루트)의 head 와 미커밋·untracked 를 포함한 tree 를 기록하고, 판정 수집 뒤와 승격 직전에 다시 계산합니다. 바뀌었으면 위 identity 모드와 같이 그 라운드의 PASS 를 버리고 수정 잡 없이 다시 리뷰합니다(마지막 반복이면 exit 1). 제어 평면(`.mam`, 잡·감사 로그 디렉터리)은 비교에서 뺍니다. 리뷰어가 ignore 되지 않는 산출물을 소스에 남기면 라운드가 stale 이 되니 산출물은 소스 밖에 둡니다. 하위 잡이 없는 루프는 이 검사를 하지 않습니다(전과 같음) — 소스에 잡마다 바뀌는 untracked 파일이 있을 수 있기 때문입니다. git 저장소가 아니거나 서브모듈이 dirty 해 지문을 만들 수 없으면 경고(`stale-review guard is off`) 후 기존 동작입니다. 루프가 끝난 **뒤**의 늦은 쓰기는 막을 수 없으며 다음 루프의 diff 에 미커밋 변경으로 드러납니다.
 

.agents/skills/multi-agent-mux-loop/scripts/run_loop.sh

@@ -39,6 +39,7 @@ LOOP_REVIEW_ONLY=0
 LOOP_REVIEW_ISOLATION=""
 LOOP_REMOTE_REVIEW=""
 LOOP_WAIT_OPEN_JOBS=0
+LOOP_JOB_TIMEOUT=""
 
 # Print usage instructions
 usage() {
@@ -53,6 +54,7 @@ usage() {
   echo "  --all-reviewer         Enforce PASS verdict from all active reviewer sessions"
   echo "  --max-loop N           Max execution-review corrective loop runs (default: 3)"
   echo "  --max-rebut N          Max rebuttal attempts per review iteration (default: 1, 0: disabled)"
+  echo "  --job-timeout <sec>    Budget of each delegated job (default: \$MAM_JOB_TIMEOUT_S or 3600)"
   echo "  --verbose              Print detailed execution timeline traces"
   echo "  --cleanup              Purge temporary job directories upon success"
   echo "  --worktree <path>      Source worktree for diff/base/head (control plane stays here)"
@@ -104,6 +106,7 @@ while [[ "$#" -gt 0 ]]; do
     --review-isolation) LOOP_REVIEW_ISOLATION="$2"; shift 2 ;;
     --remote-review) LOOP_REMOTE_REVIEW="$2"; shift 2 ;;
     --wait-open-jobs) LOOP_WAIT_OPEN_JOBS=1; shift ;;
+    --job-timeout) LOOP_JOB_TIMEOUT="${2:-}"; shift 2 ;;
     -h|--help) usage ;;
     *) echo "Unknown option: $1"; usage ;;
   esac
@@ -133,8 +136,8 @@ fi
 if [ "${MAM_LOOP_PY:-1}" = "0" ] && { [ -n "$LOOP_WORKTREE" ] || [ -n "$LOOP_BASE" ] \
      || [ "$LOOP_REVIEW_ONLY" = "1" ] || [ -n "${MAM_LOOP_WORKTREE:-}" ] \
      || [ -n "$LOOP_REVIEW_ISOLATION" ] || [ -n "$LOOP_REMOTE_REVIEW" ] \
-     || [ -n "$LOOP_PLAN_FILE" ] || [ "$LOOP_WAIT_OPEN_JOBS" = "1" ]; }; then
+     || [ -n "$LOOP_PLAN_FILE" ] || [ "$LOOP_WAIT_OPEN_JOBS" = "1" ] || [ -n "$LOOP_JOB_TIMEOUT" ]; }; then
-  echo "ERROR: --worktree/--base/--review-only/--review-isolation/--remote-review/--plan-file/--wait-open-jobs require the Python loop (MAM_LOOP_PY=1)" >&2
+  echo "ERROR: --worktree/--base/--review-only/--review-isolation/--remote-review/--plan-file/--wait-open-jobs/--job-timeout require the Python loop (MAM_LOOP_PY=1)" >&2
   exit 1
 fi
 

tests/conftest.py

@@ -391,6 +391,12 @@ while args and args[0] in ("-L", "--server", "-s", "--session"):
 if not args:
     sys.exit(0)
 
+# Issue #30: a herdr that fails every call the way a server with a
+# mismatched protocol does (exit 1, error JSON on stderr).
+if state.get("force_error_code"):
+    sys.stderr.write(json.dumps({"error": {"code": state["force_error_code"], "message": "forced"}}) + "\\n")
+    sys.exit(1)
+
 cmd1 = args[0]
 if cmd1 == "workspace":
     if len(args) < 2:
@@ -622,9 +628,10 @@ elif cmd1 == "pane":
         if len(args) < 3:
             sys.exit(1)
         pane_id = args[2]
+        detection = "--source" in args and args[args.index("--source") + 1] == "detection"
         for a_name, data in state.get("agents", {}).items():
             if data.get("pane_id") == pane_id or a_name == pane_id:
-                print(data.get("buffer", "Ready"))
+                print(data.get("detection", "") if detection else data.get("buffer", "Ready"))
                 sys.exit(0)
         for p in state.get("panes", []):
             if p.get("pane_id") == pane_id:
@@ -656,15 +663,23 @@ elif cmd1 == "pane":
         if "--pane" in args:
             pane_id = args[args.index("--pane") + 1]
         pid = 9999
+        pname = "claude"
         for name, data in state.get("agents", {}).items():
             if data.get("pane_id") == pane_id or pane_id == "w1:p1":
                 pid = data.get("pid", 9999)
+                pname = data.get("agent", "claude")
                 break
+        # Issue #30: the shape of a real herdr 0.9.1 `pane process-info` answer
+        # (tests/test_issue30_stall_watch.py REAL_PROCESS_INFO).
         res = {
             "process_info": {
-                "foreground_processes": [{"pid": pid}],
+                "foreground_process_group_id": pid,
+                "foreground_processes": [{"argv": [pname], "cmdline": pname, "cwd": os.getcwd(),
+                                          "name": pname, "pid": pid}],
+                "pane_id": pane_id or "w1:p1",
                 "shell_pid": pid
-            }
+            },
+            "type": "pane_process_info"
         }
         print(json.dumps({"result": res}))
         sys.exit(0)
@@ -936,6 +951,7 @@ elif cmd1 == "agent":
                         "agent": agent_data.get("agent", "claude"),
                         "status": agent_data.get("status", "running"),
                         "agent_status": agent_st,
+                        "state_change_seq": agent_data.get("state_change_seq", 0),
                         "interactive_ready": agent_data.get("interactive_ready", True),
                         "cwd": agent_data.get("cwd", ""),
                         "pane_id": agent_data.get("pane_id", "w1:p1"),
@@ -1010,6 +1026,10 @@ elif cmd1 == "agent":
             }))
             sys.exit(0)
         else:
+            # Issue #30: a real wait blocks until its --timeout; opt in to that
+            # (capped) so a watcher looping on waits does not spin.
+            if state.get("wait_blocks_s") and timeout_ms:
+                time.sleep(min(timeout_ms / 1000.0, float(state["wait_blocks_s"])))
             sys.stderr.write(json.dumps({
                 "error": {
                     "code": "timeout",

tests/fixtures/issue_5/phase1/help-registry.txt

@@ -11,7 +11,7 @@ positional arguments:
     get                 print one job record as JSON
     status              set a job status
     update              update a job record
-    close               close a job with deadline_timeout or cancelled
+    close               close a job with deadline_timeout, wall_clock_timeout or cancelled
     get-feedback        get the last feedback detail (completed/error) for a
                         job
     pick                claim a pending job for a session; prints id

tests/fixtures/panes/agy_idle.txt

@@ -0,0 +1,16 @@
+# captured 2026-09-25 from an idle agent, input box empty: herdr 0.9.1 `herdr agent read reviewer-agy-01` (last 15 lines)
+      • All 70 tests passed in 161.02s.
+  • Combined Test Total: 87 passed, 0 failed.
+  • Cleanup: Temporary scratch files in /tmp have been purged.
+  ──────
+  ### 4. Deliverables
+
+  • Review Report: Saved to report-final.md ending with standalone line
+  [VERDICT: PASS].
+  • Delegation Event: Published completed event with dispatch token
+  4988b0584e6b4da4af4be4d94f5f1328. Job transitioned to completed.
+
+───────────────────────────────────────────────────────────────────────────
+>
+───────────────────────────────────────────────────────────────────────────
+? for shortcuts                                     Gemini 3.8 Flash · high

tests/fixtures/panes/claude_idle.txt

@@ -0,0 +1,16 @@
+# captured 2026-09-25 from an idle agent, input box empty: herdr 0.9.1 `herdr agent read planner-reviewer-claude-01` (last 15 lines)
+  Remote and human jobs are skipped with one log line. Status reliability
+  lives in an adapter field that gates the idle judgement.
+
+  A new section maps each rule to the code that enforces it and the test
+  that pins it, and lists the four rules that cannot be enforced in code.
+  Documentation shrinks to usage lines and table rows. The tests use the
+  real wrapper, subscriber, registry, publish script and watcher with real
+  process trees, and the 14-minute end-to-end scenario runs once.
+
+✻ Cogitated for 12m 56s · done 11:04 AM
+                                    ✔ Update installed · Restart to update
+────────────────────────────────────────────────────────────────────────────
+❯
+────────────────────────────────────────────────────────────────────────────
+  ⏵⏵ bypass permissions on (shift+tab to cycle) · ← for agents

tests/fixtures/panes/codex_idle.txt

@@ -0,0 +1,16 @@
+# captured 2026-09-25 from an idle agent, input box empty: herdr 0.9.1 `herdr agent read planner-reviewer-codex-01` (last 15 lines)
+
+• Explored
+  └ Search "status"|"detail"|"completed" in b8682c96.json
+
+• 리뷰를 PASS로 완료했습니다. 추가 검증 3개와 정적 검사를 통과했으며,
+  completed 발행과 상태 반영을 확인했습니다.
+
+  최종 리포트 (.mam/jobs/b8682c96/codex-reports/report-final.md)
+
+  Worked for 5m 5s · done 2:00 AM
+
+
+› Ask Codex to do anything
+
+  gpt-6-astra high · ~/PuKi/laa/canary_projects/multi-agent-mux · Complete…

tests/fixtures/panes/grok_idle.txt

@@ -0,0 +1,16 @@
+# captured 2026-09-25 from an idle agent, input box empty: herdr 0.9.1 `herdr agent read reviewer-grok-01` (last 15 lines)
+
+     Worked for 2m54s
+                                                                           █
+
+  Help improve Grok                                     [Opt out] [Opt in]
+  Off by default. Opt-in to allow SpaceXAI to retain coding data, e.g.,
+  prompts, traces, & metrics, for training and debugging purposes. Change
+  anytime via settings.
+  Read Terms and Privacy Policy.
+
+  ╭──────────────────────────────────────────────────────────────────────╮
+  │ ❯                                                                    │
+  ╰─────────────────────────────────── Grok 4.7 (high) · always-approve ─╯
+
+  Shift+Tab:mode  │  Ctrl+.:shortcuts

tests/fixtures/panes/opencode_idle.txt

@@ -0,0 +1,16 @@
+# captured 2026-09-25 from an idle agent, input box empty: herdr 0.9.1 `herdr agent read reviewer-opencode-01` (last 15 lines)
+     유출시키고 있습니다. 하위 워크스페이스 접두사가 테스트 export와 일치하는 pane에서 실행될 때 H-21이 잘못
+     실패합니다. 이는 제 자신의 w2 pane에서 재현되었으며, 환경을 제거한 상태로 재실행하면 통과됩니다. 한 줄의
+     unset 수정으로 해결됩니다. MAM 자체 사용 모델(pane에서 에이전트 실행 테스트)에서는 이 문제가 재발
+     가능합니다.
+
+     보고서: .mam/jobs/2986037e/opencode-reports/report-final.md — VERDICT: PASS
+
+     ▣  Build · GLM 5.3 Cloud · 26m 12s
+
+  ┃
+  ┃
+  ┃                                                                                                                 ~/PuKi/laa/canary_projects/multi-
+  ┃  Build auto · GLM 5.3 Cloud Ollama                                                                              agent-mux:main
+  ╹▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀
+   /home/godopu16/PuKi/laa/canary_projects/multi-agent-mux                               629.5K  ctrl+p commands    • OpenCode 1.18.32

tests/test_a4_adapter_contract.py

@@ -720,6 +720,8 @@ def test_opencode_input_region_properties():
     assert adapter.input_prompt == '❯'
     assert adapter.input_placeholder == ''
     assert adapter.input_rule_pattern == '─{10,}'
+    # Issue #30: the input box itself is a ┃ block (herdr 0.9.1 capture).
+    assert adapter.input_block_prefix == '┃'
 
 
 def test_opencode_verify_artifact_created_at_epoch_filtering(tmp_path):

tests/test_issue18_liveness_contract.py

@@ -1030,7 +1030,7 @@ def test_a_replayed_retained_state_does_not_keep_a_silent_job_alive(tmp_path, mo
                   "--timeout", "20"])
     elapsed = time.monotonic() - started
     fake.stopped.set()
-    assert rc == 2
+    assert rc == js.RC_NO_EVENT
     assert elapsed < 8, f"the replayed state kept the silent job alive for {elapsed:.1f}s"
 
 
@@ -1286,7 +1286,9 @@ def test_loop_runner_passes_the_contract_and_remote_label(tmp_path, monkeypatch)
     r.parse_args()
     assert r.session_label("creator-claude-01") == "herdr:creator-claude-01"
     assert r.session_label("reviewer-far-01") == "remote:reviewer-far-01"
-    assert r.contract_args() == ["--heartbeat", "20", "--heartbeat-misses", "3", "--ack-timeout", "45"]
+    # Issue #33: the job budget is always passed explicitly (default 3600)
+    assert r.contract_args() == ["--timeout", "3600", "--heartbeat", "20", "--heartbeat-misses", "3",
+                                 "--ack-timeout", "45"]
     seen = []
 
     def fake_run(cmd, **_kw):

tests/test_issue30_stall_watch.py

@@ -0,0 +1,789 @@
+"""Issue #30: a stall watcher that only notifies.
+
+The watcher is a real `python -m lib_py.stall_watch` process against
+a job registered by the real wrapper (`submit --dry-run`) and moved by the real
+producer (`publish_event.py`); the broker is unreachable, so publish_event
+exits 2 after the record and events.log were written. herdr is the conftest
+mock, whose `pane process-info` answer has the shape of a real herdr 0.9.1
+answer (S1), and whose `agent read` text is a real idle-pane capture
+(tests/fixtures/panes, S14). The agent process and its children are real
+processes, so the descendant scan reads the real /proc.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+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"
+STALL_WATCH = SKILLS / "lib_py" / "stall_watch.py"
+PANES = REPO_ROOT / "tests" / "fixtures" / "panes"
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+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")
+
+UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
+
+# `herdr pane process-info --pane w1:p1` on herdr 0.9.1, 2026-09-25 (plan F11).
+REAL_PROCESS_INFO = {"result": {"process_info": {
+    "foreground_process_group_id": 11407,
+    "foreground_processes": [{"argv": ["claude"], "cmdline": "claude --dangerously-skip-permissions",
+                              "cwd": "/home/u/ws", "name": "claude", "pid": 11407}],
+    "pane_id": "w1:p1", "shell_pid": 8187}, "type": "pane_process_info"}}
+
+# The input line of each idle capture, and a synthetic "someone is typing"
+# variant of it (no real typing capture was taken: typing into another agent's
+# pane was out of scope for the capture job).
+TYPING = {"claude": (r"^❯\s*$", "❯ please continue"), "agy": (r"^>\s*$", "> please continue"),
+          "grok": (r"│ ❯ +", "│ ❯ please continue "), "opencode": (r"^  ┃\s*$", "  ┃  please continue"),
+          "codex": (r"^› Ask Codex to do anything", "› please continue")}
+
+
+def _pane(agent: str, typing: bool = False) -> str:
+    text = "".join(l for l in (PANES / f"{agent}_idle.txt").read_text().splitlines(True) if not l.startswith("# "))
+    if typing:
+        pat, rep = TYPING[agent]
+        typed = re.sub(pat, rep, text, count=1, flags=re.M)
+        assert typed != text, agent
+        return typed
+    return text
+
+
+# ---------------------------------------------------------------------------
+# S14 - the input-box facts are pinned to real captures
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("agent", sorted(TYPING))
+def test_s14_a_real_idle_capture_has_an_empty_input_box_and_typing_does_not(agent):
+    assert extract_input_region(_pane(agent), agent) == ""
+    assert extract_input_region(_pane(agent, typing=True), agent) == "please continue"
+
+
+def test_s14_rule_agents_keep_their_path_and_an_agent_without_facts_raises():
+    ruled = "old\n" + "─" * 20 + "\n❯ \n" + "─" * 20 + "\nfooter"
+    assert extract_input_region(ruled, "hermes") == ""
+    with pytest.raises(ValueError):
+        extract_input_region(ruled, "no-such-agent")
+    # no whole-pane fallback: a pane without any input line is not "empty"
+    assert extract_input_region("just a transcript\n", "codex") != ""
+    assert extract_input_region("just a transcript\n", "opencode") != ""
+
+
+# ---------------------------------------------------------------------------
+# judge - the pure decision, as a table
+# ---------------------------------------------------------------------------
+
+def _sig(**kw):
+    base = dict(agent_status="idle", idle_for=10.0, agent_pid=123, background_children=[],
+                input_empty=True, job_status="running")
+    base.update(kw)
+    return stall_watch.Signals(**base)
+
+
+@pytest.mark.parametrize("sig,st,want", [
+    (_sig(), {}, stall_watch.NOTIFY_IDLE_AND_NUDGE),
+    (_sig(agent_status="done"), {}, stall_watch.NOTIFY_IDLE_AND_NUDGE),
+    (_sig(idle_for=1.0), {}, None),
+    (_sig(background_children=["9:sleep"]), {}, None),
+    (_sig(input_empty=False), {}, None),
+    (_sig(input_empty=None), {}, None),
+    (_sig(agent_pid=None), {}, None),
+    (_sig(job_status="completed"), {}, None),
+    (_sig(), {"nudged": True}, None),
+    (_sig(), {"reliable": False}, None),
+    (_sig(agent_status="working"), {}, None),
+    (_sig(agent_status="blocked"), {}, stall_watch.NOTIFY_BLOCKED),
+    (_sig(agent_status="blocked"), {"reliable": False}, stall_watch.NOTIFY_BLOCKED),
+    (_sig(agent_status="blocked"), {"blocked_notified": True}, None),
+])
+def test_judge_table(sig, st, want):
+    assert stall_watch.judge(sig, stall_watch.WatchState(idle_s=5.0, **st)) == want
+
+
+# ---------------------------------------------------------------------------
+# harness
+# ---------------------------------------------------------------------------
+
+def _env(sandbox: 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, PYTHONPATH=str(SKILLS), MAM_JOB_ID="",
+               DELEGATE_JOB_LOGS_DIR=str(sandbox / ".mam" / "logs"))
+    env.update({k: str(v) for k, v in extra.items()})
+    return env
+
+
+def _job(env: dict, reg: Path) -> str:
+    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=env["HOME"], 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"
+    return jid
+
+
+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 _details(reg: Path, jid: str) -> list:
+    log = reg / f"{jid}.events.log"
+    if not log.exists():
+        return []
+    out = []
+    for line in log.read_text().splitlines():
+        try:
+            ev = json.loads(line)
+        except ValueError:
+            continue
+        if ev.get("event") == "progress":
+            out.append(ev.get("detail", ""))
+    return out
+
+
+def _prompts(state_file: Path) -> list:
+    return [c for c in json.loads(state_file.read_text()).get("calls", []) if c[:2] == ["agent", "prompt"]]
+
+
+def _agent(state_file: Path, name: str, agent: str, pid: int, status: str, buffer: str, detection: str = ""):
+    state = json.loads(state_file.read_text())
+    state["wait_blocks_s"] = 0.5
+    state.setdefault("agents", {})[name] = {
+        "name": name, "agent": agent, "status": "running", "agent_status": status, "pane_id": "w1:p1",
+        "workspace_id": "w1", "pid": pid, "buffer": buffer, "detection": detection}
+    state_file.write_text(json.dumps(state, indent=2))
+
+
+def _proc(script: str) -> subprocess.Popen:
+    p = subprocess.Popen(["bash", "-c", script], start_new_session=True)
+    time.sleep(0.3)  # let `exec` and the background child settle
+    return p
+
+
+def _kill(p: subprocess.Popen) -> None:
+    try:
+        os.killpg(p.pid, 9)
+    except ProcessLookupError:
+        pass
+    p.wait()
+
+
+def _watch(env: dict, reg: Path, jid: str, target: str, agent: str) -> subprocess.Popen:
+    tok = _record(reg, jid)["dispatch_token"]
+    return subprocess.Popen([sys.executable, "-m", "lib_py.stall_watch", "--job", jid,
+                             "--registry-dir", str(reg), "--target", target, "--agent", agent,
+                             "--expect-dispatch-token", tok],
+                            env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=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 _finish(env, reg, jid, w: subprocess.Popen) -> str:
+    """Complete the job for real: the watcher sees the terminal record and exits 0."""
+    _publish(env, reg, jid, "completed")
+    out, _ = w.communicate(timeout=30)
+    assert w.returncode == 0, out
+    return out
+
+
+@pytest.fixture
+def box(mam_sandbox, mock_herdr):
+    reg = mam_sandbox / ".mam" / "jobs"
+    env = _env(mam_sandbox, MAM_STALL_IDLE_S=2)
+    return mam_sandbox, mock_herdr, reg, env
+
+
+# ---------------------------------------------------------------------------
+# S1 - the mock answers in the shape of the real herdr
+# ---------------------------------------------------------------------------
+
+def test_s1_mock_process_info_has_the_real_herdr_shape(box):
+    _, state, _, env = box
+    _agent(state, "creator-claude-t", "claude", 4242, "idle", "Ready")
+    out = subprocess.run(["herdr", "pane", "process-info", "--pane", "w1:p1"], capture_output=True, text=True,
+                         env=env).stdout
+    mock = json.loads(out)
+    real, got = REAL_PROCESS_INFO["result"], mock["result"]
+    assert set(real) <= set(got)
+    assert set(real["process_info"]) <= set(got["process_info"])
+    assert set(real["process_info"]["foreground_processes"][0]) <= set(got["process_info"]["foreground_processes"][0])
+    assert got["process_info"]["foreground_processes"][0]["pid"] == 4242
+
+
+# ---------------------------------------------------------------------------
+# S2 - blocked: reported once with the dialog text, never answered
+# ---------------------------------------------------------------------------
+
+def test_s2_blocked_is_reported_once_with_the_dialog_and_never_answered(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    agent = _proc("exec sleep 300")
+    try:
+        _agent(state, "creator-claude-t", "claude", agent.pid, "blocked", _pane("claude"),
+               detection="\n Dangerous rm operation detected. Do you want to proceed?\n 1. Yes\n")
+        w = _watch(env, reg, jid, "creator-claude-t", "claude")
+        assert _until(lambda: _details(reg, jid))
+        time.sleep(2)
+        assert _details(reg, jid) == ["stalled:blocked Dangerous rm operation detected. Do you want to proceed?"]
+        calls = json.loads(state.read_text())["calls"]
+        assert not [c for c in calls if c[:2] == ["agent", "prompt"] or "send-keys" in c]
+        assert _record(reg, jid)["status"] == "running"
+        out = _finish(env, reg, jid, w)
+        assert "WARNING" in out and "blocked" in out
+    finally:
+        _kill(agent)
+
+
+# ---------------------------------------------------------------------------
+# S3/S4/S5 - idle: nudged once only without background work
+# ---------------------------------------------------------------------------
+
+def test_s3_idle_without_children_is_nudged_exactly_once(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    agent = _proc("exec sleep 300")
+    try:
+        _agent(state, "creator-claude-t", "claude", agent.pid, "idle", _pane("claude"))
+        w = _watch(env, reg, jid, "creator-claude-t", "claude")
+        assert _until(lambda: _prompts(state))
+        time.sleep(4)
+        assert _details(reg, jid) == ["stalled:idle"]
+        prompts = _prompts(state)
+        assert len(prompts) == 1
+        assert prompts[0][2] == "creator-claude-t" and prompts[0][3].startswith(f"Job {jid}: you look idle")
+        assert _record(reg, jid)["status"] == "running"
+        _finish(env, reg, jid, w)
+    finally:
+        _kill(agent)
+
+
+def test_s4_idle_with_a_background_child_is_left_alone(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    agent = _proc("sleep 300 & exec sleep 300")
+    try:
+        assert stall_watch.background_children(agent.pid)
+        _agent(state, "creator-claude-t", "claude", agent.pid, "idle", _pane("claude"))
+        w = _watch(env, reg, jid, "creator-claude-t", "claude")
+        time.sleep(6)
+        assert not [d for d in _details(reg, jid) if d.startswith("stalled:")]
+        assert not _prompts(state)
+        _finish(env, reg, jid, w)
+    finally:
+        _kill(agent)
+
+
+@pytest.mark.parametrize("agent_name,nudged", [("codex", True), ("claude", False)])
+def test_s5_the_adapters_resident_children_change_the_judgement(box, agent_name, nudged):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    agent = _proc("(exec -a codex-code-mode-host sleep 300) & exec sleep 300")
+    try:
+        _agent(state, f"reviewer-{agent_name}-t", agent_name, agent.pid, "idle", _pane(agent_name))
+        w = _watch(env, reg, jid, f"reviewer-{agent_name}-t", agent_name)
+        if nudged:
+            assert _until(lambda: _prompts(state))
+        else:
+            time.sleep(6)
+        assert len(_prompts(state)) == (1 if nudged else 0)
+        assert (_details(reg, jid) == ["stalled:idle"]) is nudged
+        _finish(env, reg, jid, w)
+    finally:
+        _kill(agent)
+
+
+# ---------------------------------------------------------------------------
+# S6 - the heartbeat timer is never background work
+# ---------------------------------------------------------------------------
+
+def test_s6_the_heartbeat_timer_is_not_a_background_child(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    tok = _record(reg, jid)["dispatch_token"]
+    publish = (f"{sys.executable} {PUBLISH} --registry-dir {reg} --job {jid} --event progress --detail x "
+               f"--expect-dispatch-token {tok} --attempts 1 --heartbeat 1; exec sleep 300")
+    agent = subprocess.Popen(["bash", "-c", publish], env=env, start_new_session=True)
+    timer = []
+    try:
+        def _timer():
+            out = subprocess.run(["pgrep", "-f", f"lib_py.delegation.heartbeat --job {jid}"],
+                                 capture_output=True, text=True).stdout.split()
+            timer[:] = [int(p) for p in out]
+            return timer
+        assert _until(_timer, 30)
+        assert _until(lambda: open(f"/proc/{agent.pid}/cmdline", "rb").read().startswith(b"sleep"), 30)
+        kids = stall_watch.background_children(agent.pid)
+        assert not [k for k in kids if int(k.split(":")[0]) in timer], kids
+    finally:
+        for pid in timer:
+            try:
+                os.kill(pid, 9)
+            except ProcessLookupError:
+                pass
+        _kill(agent)
+
+
+# ---------------------------------------------------------------------------
+# S7 - someone typing, or no input facts: no nudge, and the latter says so
+# ---------------------------------------------------------------------------
+
+def test_s7_a_typed_input_box_withholds_the_nudge(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    agent = _proc("exec sleep 300")
+    try:
+        _agent(state, "reviewer-grok-t", "grok", agent.pid, "idle", _pane("grok", typing=True))
+        w = _watch(env, reg, jid, "reviewer-grok-t", "grok")
+        time.sleep(6)
+        assert _details(reg, jid) == []
+        assert not _prompts(state)
+        out = _finish(env, reg, jid, w)
+        assert "not empty" in out
+    finally:
+        _kill(agent)
+
+
+def test_s7b_an_agent_without_input_facts_is_withheld_and_reported_once(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    agent = _proc("exec sleep 300")
+    try:
+        _agent(state, "worker-x", "noinput", agent.pid, "idle", "whatever")
+        w = _watch(env, reg, jid, "worker-x", "noinput")
+        assert _until(lambda: _details(reg, jid))
+        time.sleep(5)
+        assert _details(reg, jid) == ["stall_watch:input_check_unavailable:noinput"]
+        assert not _prompts(state)
+        out = _finish(env, reg, jid, w)
+        assert out.count("input check unavailable") == 1
+    finally:
+        _kill(agent)
+
+
+# ---------------------------------------------------------------------------
+# S8 - a failing herdr is never "no event"
+# ---------------------------------------------------------------------------
+
+def test_s8_herdr_failure_is_reported_and_the_watcher_stops_with_rc_2(box):
+    _, state, reg, env = box
+    jid = _job(env, reg)
+    st = json.loads(state.read_text())
+    st["force_error_code"] = "protocol_mismatch"
+    state.write_text(json.dumps(st))
+    env["MAM_STALL_HERDR_FAIL_S"] = "3"
+    w = _watch(env, reg, jid, "creator-claude-t", "claude")
+    out, _ = w.communicate(timeout=60)
+    assert w.returncode == 2, out
+    assert _details(reg, jid) == ["stall_watch:herdr_unavailable:protocol_mismatch"]
+    assert "herdr unavailable" in out and "WARNING" in out
+    assert _record(reg, jid)["status"] == "running"
+
+
+# ---------------------------------------------------------------------------
+# S12 - the watcher has no path that closes a job
+# ---------------------------------------------------------------------------
+
+def test_s12_the_watcher_can_only_publish_progress():
+    tree = ast.parse(STALL_WATCH.read_text())
+    names = {n.id for n in ast.walk(tree) if isinstance(n, ast.Name)}
+    attrs = {n.attr for n in ast.walk(tree) if isinstance(n, ast.Attribute)}
+    imported = {a.name for n in ast.walk(tree) if isinstance(n, (ast.Import, ast.ImportFrom)) for a in n.names}
+    froms = {n.module for n in ast.walk(tree) if isinstance(n, ast.ImportFrom)}
+    for forbidden in ("close", "update_status", "_operator_write"):
+        assert forbidden not in names | attrs | imported, forbidden
+    assert "lib_py.delegation.registry" not in froms and "registry" not in imported
+    # every `--event` in the file is followed by the literal "progress"
+    events = []
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.List, ast.Tuple)):
+            elts = node.elts
+            for i, e in enumerate(elts[:-1]):
+                if isinstance(e, ast.Constant) and e.value == "--event":
+                    events.append(getattr(elts[i + 1], "value", None))
+    assert events == ["progress"]
+    publishers = [n for n in ast.walk(tree) if isinstance(n, ast.Constant) and n.value == "lib_py.delegation.publish_event"]
+    assert len(publishers) == 1
+
+
+# ---------------------------------------------------------------------------
+# S13 - an adapter whose agent_status is unreliable gets no idle judgement
+# ---------------------------------------------------------------------------
+
+def test_s13_unreliable_agent_status_turns_idle_off_but_keeps_blocked(box):
+    _, state, reg, env = box
+    env["MAM_TEST_UNRELIABLE_AGENT"] = "claude"
+    jid = _job(env, reg)
+    agent = _proc("exec sleep 300")
+    try:
+        _agent(state, "creator-claude-t", "claude", agent.pid, "idle", _pane("claude"),
+               detection="Do you want to proceed?")
+        w = _watch(env, reg, jid, "creator-claude-t", "claude")
+        time.sleep(5)
+        assert _details(reg, jid) == [] and not _prompts(state)
+        st = json.loads(state.read_text())
+        st["agents"]["creator-claude-t"]["agent_status"] = "blocked"
+        state.write_text(json.dumps(st))
+        assert _until(lambda: _details(reg, jid))
+        assert _details(reg, jid) == ["stalled:blocked Do you want to proceed?"]
+        out = _finish(env, reg, jid, w)
+        assert "idle judgement off for claude" in out
+    finally:
+        _kill(agent)
+
+
+# ---------------------------------------------------------------------------
+# scope - the #34 rule reaches the python herdr calls
+# ---------------------------------------------------------------------------
+
+def test_scoped_env_drops_the_pane_inherited_workspace_for_name_lookups_only():
+    pane = {"HERDR_SESSION": "multi-agent-mux", "HERDR_WORKSPACE_ID": "w6", "HERDR_PANE_ID": "w6:p1"}
+    by_name = watcher.scoped_env("multi-agent-mux", environ=pane, argv=["herdr", "agent", "wait", "rev"])
+    assert "HERDR_WORKSPACE_ID" not in by_name and by_name["HERDR_PANE_ID"] == "w6:p1"
+    by_pane = watcher.scoped_env("multi-agent-mux", environ=pane, argv=["herdr", "pane", "read", "w2:p1"])
+    assert by_pane["HERDR_WORKSPACE_ID"] == "w6"
+    chosen = dict(pane, HERDR_WORKSPACE_ID="w5")
+    assert watcher.scoped_env("multi-agent-mux", environ=chosen, argv=["herdr", "agent", "get", "rev"])[
+        "HERDR_WORKSPACE_ID"] == "w5"
+    assert watcher.scoped_env(None, environ={"HERDR_WORKSPACE_ID": "w6"},
+                              argv=["herdr", "agent", "get", "rev"])["HERDR_WORKSPACE_ID"] == "w6"
+
+
+# ---------------------------------------------------------------------------
+# S9 / S10 - the real wrapper starts and stops the watcher (the #18 harness)
+# ---------------------------------------------------------------------------
+
+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()]
+
+
+@pytest.mark.parametrize("switch", ["1", "0"])
+def test_s9_the_wrapper_runs_the_watcher_for_the_jobs_life(tmp_path, mock_herdr, switch):
+    reg = tmp_path / "jobs"
+    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, **_live_broker()}, timeout_s=60)
+    jid = t18._wait_job_id(reg)
+    try:
+        time.sleep(1)
+        assert t18._publish(reg, jid, "started", env).returncode == 0
+        if switch == "1":
+            assert _until(lambda: (reg / f"{jid}.stall.out").exists() and _watchers(jid), 20), "no watcher"
+        else:
+            time.sleep(2)
+        assert t18._publish(reg, jid, "completed", env, "--detail", "done").returncode == 0
+        stdout, stderr = proc.communicate(timeout=60)
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+    assert proc.returncode == 0, stdout + stderr
+    assert _watchers(jid) == [], "the watcher outlived its job"
+    assert (reg / f"{jid}.stall.out").exists() is (switch == "1")
+    assert "stall watch log" not in stderr  # a completed job carries no watcher log
+
+
+@pytest.mark.parametrize("agent,session,why", [("human", "herdr:human", "human job"),
+                                               ("claude", "remote:far-01", "remote job")])
+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", **_live_broker()})
+    jid = t18._wait_job_id(reg)
+    try:
+        time.sleep(1)
+        assert t18._publish(reg, jid, "started", env).returncode == 0
+        time.sleep(1)
+        assert t18._publish(reg, jid, "completed", env, "--detail", "done").returncode == 0
+        stdout, stderr = proc.communicate(timeout=60)
+    finally:
+        if proc.poll() is None:
+            proc.kill()
+    assert proc.returncode == 0, stdout + stderr
+    assert f"stall watch: {why}, no local pane - stall detection off" in stderr
+    assert not (reg / f"{jid}.stall.out").exists()
+
+
+# ---------------------------------------------------------------------------
+# S11 - the loop runner relays the watcher's notices once, and keeps waiting
+# ---------------------------------------------------------------------------
+
+def test_s11_the_runner_warns_once_per_stall_notice(mam_sandbox, monkeypatch, capsys):
+    reg = mam_sandbox / ".mam" / "jobs"
+    env = _env(mam_sandbox)
+    jid = _job(env, reg)
+    for d in ("stalled:idle", "stalled:idle", "stall_watch:herdr_unavailable:protocol_mismatch", "ordinary"):
+        _publish(env, reg, jid, "progress", "--detail", d)
+    monkeypatch.setenv("WORKSPACE_ROOT", str(mam_sandbox))
+    monkeypatch.setenv("MAM_JOBS_DIR", str(reg))
+    monkeypatch.delenv("MAM_REAL_ROOT", raising=False)
+    from lib_py.loop import runner as lr
+    r = lr.LoopRunner(["--creator", "c", "--task", "t"])
+    r.parse_args()
+    import threading
+    threading.Timer(2.5, lambda: _publish(env, reg, jid, "completed")).start()
+    assert r.wait_for_job(jid, max_wait=30, check_interval=0.3) is True
+    out = capsys.readouterr()
+    text = out.out + out.err
+    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=40)
+    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'])
+        # S must exceed the 30 s re-judgement interval: with the old code the nudge came
+        # ~30 s after the missed transition, which only fails the assertion when S > 30
+        # (review ad3d8206; S=10 passed on the defective code too).
+        assert idle_time >= 40, {'continuous_idle_seconds': idle_time, 'threshold': 40}
+    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"
+
+
+
+# Review ad3d8206 (planner-reviewer-codex-01): a prompt glyph inside the draft
+# is draft text, not the start of the input box (the reviewer's reproduction).
+@pytest.mark.parametrize("draft", [False, True])
+def test_r1b_a_draft_containing_the_prompt_glyph_is_not_empty(mam_sandbox, mock_herdr, draft):
+    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 draft:
+            pane = pane.replace("› Ask Codex to do anything", "› Explain this terminal prompt:\n  ›")
+        _agent(mock_herdr, "reviewer-codex-t", "codex", agent.pid, "idle", pane)
+        w = _watch(env, reg, jid, "reviewer-codex-t", "codex")
+        if draft:
+            time.sleep(5)
+        else:
+            assert _until(lambda: _prompts(mock_herdr))
+        prompts = _prompts(mock_herdr)
+        _finish(env, reg, jid, w)
+        assert len(prompts) == (0 if draft else 1), prompts
+    finally:
+        if w is not None and w.poll() is None:
+            w.terminate()
+            w.communicate(timeout=10)
+        _kill(agent)
+
+
+def test_r1b_the_extractor_strips_the_prompt_from_the_first_line_only():
+    from lib_py.agents.input_region import extract_input_region
+    idle = _pane("codex")
+    for draft in ("› Explain this terminal prompt:\n  ›", "›\n  ›", "› Ask Codex to do anything\n  more"):
+        assert extract_input_region(idle.replace("› Ask Codex to do anything", draft), "codex") != "", draft
+    typed = re.sub(r"^❯\s*$", "❯ x\n  ❯", _pane("claude"), count=1, flags=re.M)
+    assert typed != _pane("claude") and extract_input_region(typed, "claude") != ""

tests/test_issue33_budget.py

@@ -0,0 +1,358 @@
+"""Issue #33: the job budget travels, the wait caps follow it, and the two
+clocks that close a job carry different labels.
+
+- B1: the real wrapper takes the budget from --timeout > $MAM_JOB_TIMEOUT_S > 3600.
+- B2: the loop runner derives its wait caps from the budget (+300 s) and hands
+  the wrapper that budget explicitly; both hops run for real.
+- B3: the real wrapper + subscriber label a spent budget `wall_clock_timeout`
+  (subscriber rc 4) and silence `deadline_timeout` (rc 2); caller rc is 2 for
+  both. These need the live broker from the environment (like test_issue6 t1_1).
+- B4: the remote runner's two deadline wordings lead with the new label.
+- B5: the sub-job table files `wall_clock_timeout` closes as unconfirmed.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import signal
+import subprocess
+import sys
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+SKILLS = REPO_ROOT / ".agents" / "skills"
+WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+REGISTRY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
+PUBLISH = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
+
+if str(SKILLS) not in sys.path:
+    sys.path.insert(0, str(SKILLS))
+
+UNREACHABLE = {"MQTT_BROKER": "127.0.0.1", "MQTT_PORT": "9", "MQTT_TLS": "0"}
+
+
+def _env(home: Path, extra=None, broker: bool = False) -> dict:
+    env = dict(os.environ)
+    env["DELEGATE_JOB_PYTHON"] = sys.executable
+    env["AGENT_PYTHON_BIN"] = sys.executable
+    env["HOME"] = str(home)
+    env["WORKSPACE_ROOT"] = str(home)
+    env["PYTHONPATH"] = str(SKILLS)
+    env["DELEGATE_JOB_LOGS_DIR"] = str(home / "logs")
+    for k in ("MAM_JOBS_DIR", "MAM_REAL_ROOT", "MAM_JOB_ID", "MAM_JOB_TIMEOUT_S", "MAM_CHILD_WAIT_S",
+              "MAM_WAIT_GRACE_S"):
+        env.pop(k, None)
+    if not broker:
+        env["MAM_ENV_FILE"] = str(home / ".mam.env.absent")
+        for k in ("MQTT_BROKER", "MQTT_HOST", "MQTT_PORT", "MQTT_TLS", "MQTT_USERNAME", "MQTT_PASSWORD"):
+            env.pop(k, None)
+        env.update(UNREACHABLE)
+    if extra:
+        env.update({k: str(v) for k, v in extra.items()})
+    return env
+
+
+def _dry_submit(env: dict, reg: Path, *extra: str) -> subprocess.CompletedProcess:
+    return subprocess.run(["bash", str(WRAPPER), "submit", "--agent", "human", "--dry-run",
+                           "--registry-dir", str(reg), "--prompt", "p", *extra],
+                          capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=120)
+
+
+def _jid(res: subprocess.CompletedProcess) -> str:
+    assert res.returncode == 0, res.stdout + res.stderr
+    m = re.search(r"registered job:\s*(\S+)", res.stdout)
+    assert m, res.stdout
+    return m.group(1)
+
+
+def _record(reg: Path, jid: str) -> dict:
+    return json.loads((reg / f"{jid}.json").read_text(encoding="utf-8"))
+
+
+def _events(reg: Path, jid: str) -> list:
+    p = reg / f"{jid}.events.log"
+    return [json.loads(l) for l in p.read_text().splitlines() if l.strip()] if p.exists() else []
+
+
+def _publish(env: dict, reg: Path, jid: str, event: str, *extra: str) -> subprocess.CompletedProcess:
+    tok = _record(reg, jid)["dispatch_token"]
+    res = subprocess.run([sys.executable, str(PUBLISH), "--registry-dir", str(reg), "--job", jid,
+                          "--event", event, "--expect-dispatch-token", tok, *extra],
+                         capture_output=True, text=True, env=env, timeout=120)
+    assert res.returncode in (0, 2), res.stderr
+    return res
+
+
+def _kill_timer(reg: Path, jid: str) -> None:
+    try:
+        os.kill(int((reg / f"{jid}.heartbeat.pid").read_text().strip()), signal.SIGKILL)
+    except (OSError, ValueError):
+        pass
+
+
+# ---------------------------------------------------------------------------
+# B1 - the wrapper's budget: --timeout > MAM_JOB_TIMEOUT_S > 3600
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("env_val,flag,expected,warns", [
+    (None, None, 3600, False),
+    ("7", None, 7, False),
+    ("7", "9", 9, False),
+    ("abc", None, 3600, True),
+])
+def test_b1_the_wrapper_takes_the_budget_from_flag_env_or_default(tmp_path, env_val, flag, expected, warns):
+    env = _env(tmp_path, {"MAM_JOB_TIMEOUT_S": env_val} if env_val else None)
+    res = _dry_submit(env, tmp_path / "reg", *(["--timeout", flag] if flag else []))
+    assert _record(tmp_path / "reg", _jid(res))["timeout_sec"] == expected
+    assert ("MAM_JOB_TIMEOUT_S=abc ignored" in res.stderr) is warns, res.stderr
+
+
+# ---------------------------------------------------------------------------
+# B2 - the runner: budget -> wait caps, and the budget handed to the wrapper
+# ---------------------------------------------------------------------------
+
+def _runner(tmp_path, monkeypatch, *argv, **env):
+    monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
+    monkeypatch.setenv("MAM_JOBS_DIR", str(tmp_path / "reg"))
+    monkeypatch.setenv("DELEGATE_JOB_LOGS_DIR", str(tmp_path / "logs"))
+    for k in ("MAM_REAL_ROOT", "MAM_JOB_TIMEOUT_S", "MAM_CHILD_WAIT_S", "MAM_WAIT_GRACE_S", "MAM_JOB_TREE_MAX"):
+        monkeypatch.delenv(k, raising=False)
+    for k, v in env.items():
+        monkeypatch.setenv(k, str(v))
+    from lib_py.loop import runner as lr
+    r = lr.LoopRunner(["--creator", "c", "--task", "t", *argv])
+    r.parse_args()
+    return r
+
+
+def test_b2_the_wait_caps_are_the_budget_plus_300(tmp_path, monkeypatch):
+    assert _runner(tmp_path, monkeypatch, "--job-timeout", "10").job_wait_s == 310
+    assert _runner(tmp_path, monkeypatch, MAM_JOB_TIMEOUT_S=20).job_wait_s == 320
+    assert _runner(tmp_path, monkeypatch, "--job-timeout", "10", MAM_JOB_TIMEOUT_S=20).job_wait_s == 310
+    r = _runner(tmp_path, monkeypatch)
+    assert (r.job_budget, r.job_wait_s) == (3600, 3900)  # no option: today's numbers
+    assert "--job-timeout" in "\n".join(_usage_lines(r))
+
+
+def _usage_lines(r):
+    import io, contextlib
+    buf = io.StringIO()
+    with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf):
+        try:
+            r.print_usage()
+        except SystemExit:
+            pass
+    return buf.getvalue().splitlines()
+
+
+def test_b2_an_invalid_job_timeout_is_refused(tmp_path, monkeypatch):
+    with pytest.raises(SystemExit):
+        _runner(tmp_path, monkeypatch, "--job-timeout", "0")
+
+
+def test_b2_the_child_wait_defaults_to_the_budget_and_an_explicit_value_wins(tmp_path, monkeypatch, capsys):
+    env = _env(tmp_path)
+    reg = tmp_path / "reg"
+    p = _jid(_dry_submit(env, reg))
+    _jid(_dry_submit(env, reg, "--parent-job", p))
+
+    start = time.time()
+    r = _runner(tmp_path, monkeypatch, "--job-timeout", "1", MAM_WAIT_GRACE_S=1)
+    assert r.wait_for_descendants(p, interval=0.5) is False
+    assert 2 <= time.time() - start < 8
+    assert "ended after 2s (MAM_CHILD_WAIT_S default: job budget 1 + 1)" in capsys.readouterr().out
+
+    start = time.time()
+    r = _runner(tmp_path, monkeypatch, "--job-timeout", "600", MAM_CHILD_WAIT_S=2)
+    assert r.wait_for_descendants(p, interval=0.5) is False
+    assert 2 <= time.time() - start < 8
+    assert "ended after 2s (MAM_CHILD_WAIT_S)" in capsys.readouterr().out
+
+
+def test_b2_the_runner_hands_the_wrapper_its_budget_and_the_wrapper_registers_it(tmp_path, monkeypatch):
+    # hop 1: the runner execs the wrapper with --timeout (a recording wrapper at the control root)
+    rec = tmp_path / "rec.args"
+    w = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+    w.parent.mkdir(parents=True)
+    w.write_text(f'#!/usr/bin/env bash\nprintf "%s\\n" "$@" > "{rec}"\necho "registered job: fake33"\n')
+    w.chmod(0o755)
+    r = _runner(tmp_path, monkeypatch, "--job-timeout", "10")
+    assert r.delegate_job("c", "claude", "Worker", "task") == "fake33"
+    args = rec.read_text().splitlines()
+    assert args[args.index("--timeout") + 1] == "10"
+    # env only: still passed explicitly
+    r = _runner(tmp_path, monkeypatch, MAM_JOB_TIMEOUT_S=20)
+    assert r.contract_args()[:2] == ["--timeout", "20"]
+
+    # hop 2: the real wrapper registers what the runner hands it
+    reg = tmp_path / "reg2"
+    res = _dry_submit(_env(tmp_path), reg, *_runner(tmp_path, monkeypatch, "--job-timeout", "10").contract_args())
+    assert _record(reg, _jid(res))["timeout_sec"] == 10
+
+
+# ---------------------------------------------------------------------------
+# B3 - the real wrapper + subscriber: which clock closed the job
+# ---------------------------------------------------------------------------
+
+def _run_wrapper_bg(env: dict, reg: Path, *extra: str) -> subprocess.Popen:
+    return subprocess.Popen(["bash", str(WRAPPER), "submit", "--agent", "human", "--registry-dir", str(reg),
+                             "--workdir", env["HOME"], "--prompt", "b3", *extra],
+                            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, cwd=env["HOME"])
+
+
+def _wait_subscribed(reg: Path, timeout: float = 30) -> str:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        for f in reg.glob("*.subscriber.out"):
+            if "SUBSCRIBED" in f.read_text(errors="replace"):
+                return f.name.split(".")[0]
+        time.sleep(0.1)
+    raise AssertionError("the subscriber never subscribed (is the broker up?)")
+
+
+def _last_error(reg: Path, jid: str) -> str:
+    errs = [e for e in _events(reg, jid) if e.get("event") == "error"]
+    assert errs, _events(reg, jid)
+    return errs[-1].get("detail", "")
+
+
+def test_b3_wall_clock_before_idle_is_labelled_wall_clock_timeout(tmp_path):
+    env = _env(tmp_path, broker=True)
+    reg = tmp_path / "jobs"
+    proc = _run_wrapper_bg(env, reg, "--timeout", "3", "--idle-timeout", "30")
+    out, err = proc.communicate(timeout=90)
+    assert proc.returncode == 2, out + err
+    jid = next(reg.glob("*.json")).stem
+    assert _record(reg, jid)["status"] == "error"
+    assert _last_error(reg, jid) == "wall_clock_timeout"
+    assert "wall-clock timeout" in (reg / f"{jid}.subscriber.out").read_text()
+    assert "closed by wall_clock clock" in err
+
+
+def test_b3_silence_is_still_labelled_deadline_timeout(tmp_path):
+    env = _env(tmp_path, broker=True)
+    reg = tmp_path / "jobs"
+    proc = _run_wrapper_bg(env, reg, "--timeout", "30", "--idle-timeout", "2")
+    try:
+        jid = _wait_subscribed(reg)
+        _publish(env, reg, jid, "started")  # then nothing: silence for K x N
+        out, err = proc.communicate(timeout=90)
+    finally:
+        proc.kill()
+    assert proc.returncode == 2, out + err
+    assert _record(reg, jid)["status"] == "error"
+    assert _last_error(reg, jid) == "deadline_timeout"
+    assert "idle timeout" in (reg / f"{jid}.subscriber.out").read_text()
+    assert "closed by no_event clock" in err
+
+
+def test_b3_a_heartbeating_job_that_spends_its_budget_is_wall_clock_timeout(tmp_path):
+    """The #33 motivating case: a live agent (heartbeats keep coming) whose
+    work outlasts the budget was reported as `deadline_timeout`."""
+    env = _env(tmp_path, broker=True)
+    reg = tmp_path / "jobs"
+    proc = _run_wrapper_bg(env, reg, "--timeout", "6", "--idle-timeout", "3")
+    jid = None
+    try:
+        jid = _wait_subscribed(reg)
+        _publish(env, reg, jid, "started", "--heartbeat", "1")
+        out, err = proc.communicate(timeout=90)
+    finally:
+        proc.kill()
+        if jid:
+            _kill_timer(reg, jid)
+    assert proc.returncode == 2, out + err
+    assert _last_error(reg, jid) == "wall_clock_timeout"
+    assert "wall-clock timeout" in (reg / f"{jid}.subscriber.out").read_text()
+
+
+# ---------------------------------------------------------------------------
+# B4 - the remote runner's deadlines are wall clocks
+# ---------------------------------------------------------------------------
+
+def _dispatch(timeout_sec=1) -> dict:
+    return {"job_id": uuid.uuid4().hex[:8], "brief": "b", "dispatch_token": "tok-" + uuid.uuid4().hex,
+            "attempt_id": 1, "auth_token": None, "timeout_sec": timeout_sec, "heartbeat_sec": 30,
+            "heartbeat_misses": 4, "ack_timeout_sec": 5}
+
+
+@pytest.mark.parametrize("cmd,prefix", [("sleep 5", "wall_clock_timeout: agent command exceeded timeout_sec 1"),
+                                        (None, "wall_clock_timeout (remote runner)")])
+def test_b4_the_remote_runner_labels_its_deadline_wall_clock_timeout(tmp_path, monkeypatch, cmd, prefix):
+    from lib_py.delegation import remote
+    for k, v in _env(tmp_path).items():
+        monkeypatch.setenv(k, v)
+    monkeypatch.chdir(tmp_path)
+    reg = tmp_path / "jobs"
+    d = _dispatch()
+    try:
+        remote.handle_dispatch(d, str(reg), "s", cmd)
+        assert _record(reg, d["job_id"])["status"] == "error"
+        assert _last_error(reg, d["job_id"]).startswith(prefix)
+    finally:
+        _kill_timer(reg, d["job_id"])
+
+
+# ---------------------------------------------------------------------------
+# B5 - the sub-job table: a wall-clock close is unconfirmed, like a deadline close
+# ---------------------------------------------------------------------------
+
+@pytest.mark.parametrize("request_kind,detail", [
+    ("wall_clock_timeout", "wall_clock_timeout"),
+    ("wall_clock_timeout", "wall_clock_timeout: agent command exceeded timeout_sec 1"),
+    ("deadline_timeout", "deadline_timeout"),
+])
+def test_b5_a_wall_clock_close_is_closed_unconfirmed(tmp_path, monkeypatch, request_kind, detail):
+    env = _env(tmp_path)
+    reg = tmp_path / "reg"
+    p = _jid(_dry_submit(env, reg))
+    c = _jid(_dry_submit(env, reg, "--parent-job", p))
+    tok = _record(reg, c)["dispatch_token"]
+    res = subprocess.run([sys.executable, str(REGISTRY), "--registry-dir", str(reg), "close", "--job", c,
+                          "--request", request_kind, "--expect-dispatch-token", tok],
+                         capture_output=True, text=True, env=env, timeout=120)
+    assert res.returncode == 0, res.stderr
+    assert _record(reg, c)["status"] == "error"
+    _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)
+
+
+
+# Issue #33 follow-up (job 60da22f6): `delegate-job wait` passes the subscriber's
+# code through; a spent budget stays rc 2 for its callers, as before the split.
+@pytest.mark.parametrize("state,rc", [("running", 2), ("completed", 0)])
+def test_b7_wait_keeps_a_spent_budget_at_rc_2(tmp_path, state, rc):
+    env = _env(tmp_path, broker=True)
+    reg = tmp_path / "reg"
+    jid = _jid(_dry_submit(env, reg))
+    if state == "completed":
+        _publish(env, reg, jid, "completed")
+    res = subprocess.run(["bash", str(WRAPPER), "wait", "--job", jid, "--registry-dir", str(reg), "--timeout", "2"],
+                         capture_output=True, text=True, env=env, cwd=env["HOME"], timeout=60)
+    assert res.returncode == rc, res.stdout + res.stderr

tests/test_issue5_phase1_delegation.py

@@ -359,9 +359,13 @@ def test_job_subscriber_cli_contract_snapshot(mam_sandbox):
     assert job_subscriber.TERMINAL_EVENTS == ("completed", "error")
     src = Path(job_subscriber.__file__).read_text(encoding="utf-8")
     for doc in ("  0  all watched jobs reached", "  1  any watched job reached",
-                "  2  timed out"):
+                "  2  no event for the idle window", "  4  the wall-clock budget"):
         assert doc in src, doc
-    assert 'return 3' in src, "fatal-error exit code 3 must survive"
+    # Issue #33: codes are named; their values are the frozen contract.
+    assert (job_subscriber.RC_COMPLETED, job_subscriber.RC_ERROR, job_subscriber.RC_NO_EVENT,
+            job_subscriber.RC_BROKER_UNAVAILABLE, job_subscriber.RC_WALL_CLOCK) == (0, 1, 2, 3, 4)
+    assert 'return RC_BROKER_UNAVAILABLE' in src, "fatal-error exit code 3 must survive"
+    assert not re.search(r"^\s+return [0-9]+\s*(#.*)?$", src, re.M), "exit codes are returned by name"
 
 
 # ==============================================================================
@@ -378,17 +382,25 @@ def test_event_to_status_mapping_is_frozen(mam_sandbox):
     }
 
 
-def test_wrapper_exit_code_mapping_literals_are_frozen():
+def test_wrapper_exit_code_mapping_literals_are_frozen(mam_sandbox):
     wrapper = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
     text = wrapper.read_text(encoding="utf-8")
     for literal in (
-        'if [[ $sub_rc -eq 0 ]]; then',
+        'if [[ $sub_rc -eq $SUB_RC_COMPLETED ]]; then',
-        'elif [[ $sub_rc -eq 1 ]]; then',
+        'elif [[ $sub_rc -eq $SUB_RC_ERROR ]]; then',
-        'elif [[ $sub_rc -eq 3 ]]; then',
+        'elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then',
+        '[ "$sub_rc" = "$SUB_RC_WALL_CLOCK" ] && clock="wall"',
         'job_status="completed"',
         'job_status="error"',
     ):
         assert literal in text, literal
+    # Issue #33: the wrapper's SUB_RC_* mirror job_subscriber's RC_* exactly,
+    # and the wrapper compares subscriber codes only by those names.
+    job_subscriber = _import_delegation(mam_sandbox, "job_subscriber")
+    mirror = dict(re.findall(r"\bSUB_RC_([A-Z_]+)=([0-9]+)", text))
+    assert mirror == {name[3:]: str(getattr(job_subscriber, name))
+                      for name in dir(job_subscriber) if name.startswith("RC_")}, mirror
+    assert not re.search(r"\$(sub_rc|sub_exit)\"? +(-eq|-ne|=) +\"?[0-9]", text), "subscriber codes compared by value"
 
 
 def test_reconcile_terminal_mutation_literals_are_frozen():

tests/test_tier1_unit.py

@@ -721,7 +721,7 @@ def test_g5_subscriber_disk_fallback_on_broker_down_exits_0(mam_sandbox, monkeyp
         "--timeout", "5",
         "--idle-timeout", "2"
     ])
-    assert rc == 0
+    assert rc == job_subscriber.RC_COMPLETED
 
 
 def test_g6_subscriber_disk_fallback_outputs_source_tag(mam_sandbox, capsys, monkeypatch):
@@ -744,7 +744,7 @@ def test_g6_subscriber_disk_fallback_outputs_source_tag(mam_sandbox, capsys, mon
         "--job", "g6test01",
         "--timeout", "5"
     ])
-    assert rc == 0
+    assert rc == job_subscriber.RC_COMPLETED
     captured = capsys.readouterr()
     assert "disk-fallback" in captured.out
 
@@ -769,7 +769,7 @@ def test_g7_subscriber_disk_fallback_error_status_exits_1(mam_sandbox, monkeypat
         "--job", "g7test01",
         "--timeout", "5"
     ])
-    assert rc == 1
+    assert rc == job_subscriber.RC_ERROR
 
 
 def test_g8_subscriber_wait_any_does_not_exit_early_if_partial_pending(mam_sandbox, monkeypatch):
@@ -806,13 +806,15 @@ def test_g8_subscriber_wait_any_does_not_exit_early_if_partial_pending(mam_sandb
     monkeypatch.setattr(job_subscriber, "make_client", lambda *args, **kwargs: FakeClient())
     monkeypatch.setattr(mqtt_common, "with_retry", lambda fn, **kwargs: fn)
 
+    # The wall clock (0.5 s) runs out well before the idle window (30 s), so
+    # which timeout ends the wait is decided, not a race (Issue #33 rc split).
     rc = job_subscriber.main([
         "--registry-dir", reg_dir,
         "--wait-any",
         "--timeout", "0.5",
-        "--idle-timeout", "0.5"
+        "--idle-timeout", "30"
     ])
-    assert rc == 2, "Must timeout waiting for incomplete job2 rather than exiting 0"
+    assert rc == job_subscriber.RC_WALL_CLOCK, "Must timeout waiting for incomplete job2 rather than exiting 0"
 
 
 def test_g9_subscriber_broker_down_no_disk_terminal_exits_3(mam_sandbox, monkeypatch):
@@ -835,14 +837,14 @@ def test_g9_subscriber_broker_down_no_disk_terminal_exits_3(mam_sandbox, monkeyp
         "--job", "g9test01",
         "--timeout", "5"
     ])
-    assert rc == 3, f"Expected rc=3 on infrastructure broker down without disk terminal, got {rc}"
+    assert rc == job_subscriber.RC_BROKER_UNAVAILABLE, f"Expected rc=3 on infrastructure broker down without disk terminal, got {rc}"
 
 
 def test_g10_delegate_job_rc3_not_mistaken_for_error(mam_sandbox):
     """G-10: multi-agent-mux-delegate-job maps rc=3 to broker_unavailable and checks disk status."""
     delegate_script = mam_sandbox / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
     content = delegate_script.read_text()
-    assert "elif [[ $sub_rc -eq 3 ]]; then" in content
+    assert "elif [[ $sub_rc -eq $SUB_RC_BROKER_UNAVAILABLE ]]; then" in content
     assert 'job_status="broker_unavailable"' in content