"""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 # Only S6 needs Linux (/proc); the rest run on macOS too (PR #41 review M8). PROC = 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() # The real watcher's main() in a process whose `agent` adapter reports an # agent_status that cannot drive the idle judgement (S13) - the production # code has no switch for it. _UNRELIABLE_MAIN = ("import sys; from lib_py.agents.registry import get_adapter; " "type(get_adapter(sys.argv[1])).agent_status_reliable = property(lambda self: False); " "import lib_py.stall_watch as sw; sys.exit(sw.main(sys.argv[2:]))") def _watch(env: dict, reg: Path, jid: str, target: str, agent: str, unreliable: bool = False) -> subprocess.Popen: tok = _record(reg, jid)["dispatch_token"] argv = ["--job", jid, "--registry-dir", str(reg), "--target", target, "--agent", agent, "--expect-dispatch-token", tok] entry = ["-c", _UNRELIABLE_MAIN, agent] if unreliable else ["-m", "lib_py.stall_watch"] return subprocess.Popen([sys.executable, *entry, *argv], 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 # --------------------------------------------------------------------------- @PROC 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 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", unreliable=True) 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) def test_s13_the_unreliable_case_needs_no_switch_in_the_product(): """PR #41 M7: the test sets the adapter up (_UNRELIABLE_MAIN); the watcher reads no environment variable to pretend an adapter is unreliable.""" assert "MAM_TEST_" not in STALL_WATCH.read_text() # --------------------------------------------------------------------------- # 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 from _local_mqtt import local_broker_env # noqa: E402 (Issue #26: the session's in-process broker) 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, **local_broker_env()}, 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", **local_broker_env()}) 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(local_broker_env()) 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 # time.time(), not monotonic: these stamps are compared across processes, # and macOS gives each process its own monotonic reference. 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.time() 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.time() 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.time() 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") != ""