GodopuandClaude Opus 5.5
c252ac4d54
stall_watch read MAM_TEST_UNRELIABLE_AGENT to pretend an adapter's
agent_status was unreliable (PR #41 M7). The branch is gone. S13 now runs
the real stall_watch.main() in a process where the adapter itself reports
agent_status_reliable = False (_UNRELIABLE_MAIN), and checks the same
things as before: no idle notice or nudge, the blocked notice, and the
"idle judgement off" log line. A static check keeps MAM_TEST_ out of the
watcher (fails at 0051fd6). Without the seam S13 fails on this tree.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
803 lines
35 KiB
Python
803 lines
35 KiB
Python
"""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") != ""
|