diff --git a/.agents/skills/lib_py/agents/__main__.py b/.agents/skills/lib_py/agents/__main__.py index 0796dfb..9651a65 100644 --- a/.agents/skills/lib_py/agents/__main__.py +++ b/.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)}") diff --git a/.agents/skills/lib_py/agents/adapters/codex.py b/.agents/skills/lib_py/agents/adapters/codex.py index 8b688fc..183b7d1 100644 --- a/.agents/skills/lib_py/agents/adapters/codex.py +++ b/.agents/skills/lib_py/agents/adapters/codex.py @@ -42,10 +42,20 @@ 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 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" diff --git a/.agents/skills/lib_py/agents/adapters/grok.py b/.agents/skills/lib_py/agents/adapters/grok.py index 055a4ab..a9a0503 100644 --- a/.agents/skills/lib_py/agents/adapters/grok.py +++ b/.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='') diff --git a/.agents/skills/lib_py/agents/adapters/opencode.py b/.agents/skills/lib_py/agents/adapters/opencode.py index c1b6272..8338ca2 100644 --- a/.agents/skills/lib_py/agents/adapters/opencode.py +++ b/.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 · ` 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: diff --git a/.agents/skills/lib_py/agents/base.py b/.agents/skills/lib_py/agents/base.py index 75d47ef..6e3ca1e 100644 --- a/.agents/skills/lib_py/agents/base.py +++ b/.agents/skills/lib_py/agents/base.py @@ -152,6 +152,34 @@ 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 inside such a block (#30).""" + 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 diff --git a/.agents/skills/lib_py/agents/input_region.py b/.agents/skills/lib_py/agents/input_region.py index e627bb9..939fdde 100644 --- a/.agents/skills/lib_py/agents/input_region.py +++ b/.agents/skills/lib_py/agents/input_region.py @@ -5,33 +5,60 @@ 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. - If rules are absent, falls back to full pane_text to support mocks and unruled panes. - Strips leading prompt tokens ('>', '❯') and placeholders ('Ask anything...'). - Raises ValueError for unsupported agents (like hermes with no input area facts). + Extracts the input area text of a pane, using only the agent adapter's facts + (issue #30, measured on herdr 0.9.1 captures in tests/fixtures/panes): + + 1. an `input_block_prefix` -> the last contiguous block of lines starting + with it, prefix removed, cut at the first `input_stop_pattern` line; + 2. two or more `input_rule_pattern` rules -> the lines between the last two; + 3. an `input_prompt` -> the last line starting with the prompt. + + Then `input_frame_chars`, the prompt and the placeholder are stripped. There + is no whole-pane fallback: a pane with none of these shapes has no input + 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)] + prompt = adapter.input_prompt + region_lines = None - 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 + 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: + starts = [line for line in lines if line.lstrip().startswith(prompt)] + # No input line at all: report the whole tail as "not empty" (unknown). + region_lines = starts[-1:] if starts else [pane_text or '?'] + + frame = adapter.input_frame_chars + if frame: + region_lines = [line.strip().strip(frame) for line in region_lines] region_text = "\n".join(region_lines) - - prompt = adapter.input_prompt - if prompt: - region_text = re.sub(r'^\s*' + re.escape(prompt) + r'\s*', '', region_text, flags=re.MULTILINE) + region_text = re.sub(r'^\s*' + re.escape(prompt) + r'\s*', '', region_text, flags=re.MULTILINE) placeholder = adapter.input_placeholder if placeholder: diff --git a/.agents/skills/lib_py/loop/runner.py b/.agents/skills/lib_py/loop/runner.py index 8d1081a..7fee63d 100644 --- a/.agents/skills/lib_py/loop/runner.py +++ b/.agents/skills/lib_py/loop/runner.py @@ -106,6 +106,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 = "" @@ -637,6 +638,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.") @@ -650,6 +652,30 @@ class LoopRunner: log_error(f"Job '{job_id}' timed out after {max_wait}s.") return False + 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) # ------------------------------------------------------------------ diff --git a/.agents/skills/lib_py/loop/watcher.py b/.agents/skills/lib_py/loop/watcher.py index f318f0e..922435e 100644 --- a/.agents/skills/lib_py/loop/watcher.py +++ b/.agents/skills/lib_py/loop/watcher.py @@ -74,21 +74,39 @@ def caller_server(environ=None): return "" -def scoped_env(session=None, environ=None): - """The environment for a wait aimed at `session`. +def _names_agent(argv) -> bool: + """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 - and dropped. No pane and same-server keep every value. + +def scoped_env(session=None, environ=None, argv=None): + """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: - return dict(src) - return dict((k, v) for k, v in src.items() if k not in CALLER_SCOPE_VARS) + if caller and target and target != "default" and caller != target: + 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)) diff --git a/.agents/skills/lib_py/stall_watch.py b/.agents/skills/lib_py/stall_watch.py new file mode 100644 index 0000000..7008b8a --- /dev/null +++ b/.agents/skills/lib_py/stall_watch.py @@ -0,0 +1,367 @@ +"""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 ` 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:`, 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: + 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): + 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 + if sig.agent_status not in IDLE_STATES or not st.reliable or st.nudged: + return + sig.idle_for = time.monotonic() - idle_since + if sig.idle_for < st.idle_s: + return + seq = sig.state_change_seq + self.src.collect(sig) + if seq is not None and sig.state_change_seq is not None and sig.state_change_seq != seq: + return # a transition we did not see: the next wait re-observes + 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)) + + +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()) diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 66c08aa..70780cc 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -365,6 +365,7 @@ EOF 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 @@ -379,6 +380,7 @@ EOF local sub_rc=0 wait "$sub_pid" || sub_rc=$? + stop_stall_watch echo "subscriber output:"; cat "$logf" || true local job_status @@ -548,10 +550,12 @@ EOF 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 @@ -672,6 +676,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 @@ -818,6 +863,11 @@ finalize_unfinished() { 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 diff --git a/tests/conftest.py b/tests/conftest.py index 951dfb4..fe597ee 100644 --- a/tests/conftest.py +++ b/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", diff --git a/tests/fixtures/panes/agy_idle.txt b/tests/fixtures/panes/agy_idle.txt new file mode 100644 index 0000000..c9fefba --- /dev/null +++ b/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 diff --git a/tests/fixtures/panes/claude_idle.txt b/tests/fixtures/panes/claude_idle.txt new file mode 100644 index 0000000..a97e20f --- /dev/null +++ b/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 diff --git a/tests/fixtures/panes/codex_idle.txt b/tests/fixtures/panes/codex_idle.txt new file mode 100644 index 0000000..2193c1a --- /dev/null +++ b/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… diff --git a/tests/fixtures/panes/grok_idle.txt b/tests/fixtures/panes/grok_idle.txt new file mode 100644 index 0000000..b13f238 --- /dev/null +++ b/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 diff --git a/tests/fixtures/panes/opencode_idle.txt b/tests/fixtures/panes/opencode_idle.txt new file mode 100644 index 0000000..fe1d3f3 --- /dev/null +++ b/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 diff --git a/tests/test_a4_adapter_contract.py b/tests/test_a4_adapter_contract.py index 1a7d0d7..1003284 100644 --- a/tests/test_a4_adapter_contract.py +++ b/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): diff --git a/tests/test_issue30_stall_watch.py b/tests/test_issue30_stall_watch.py new file mode 100644 index 0000000..a1b6d4b --- /dev/null +++ b/tests/test_issue30_stall_watch.py @@ -0,0 +1,565 @@ +"""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 + + +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}, 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"}) + 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