Files
multi-agent-mux/tests/test_issue14_delivery_ack.py
T
GodopuandClaude Opus 5 952657efb4 fix(issue-9,14): bound the R1 grace by the wall clock and keep other ls failures unknown
Round-2 review of d5a191c (planner-reviewer-claude-01, planner-reviewer-codex-01).

- delegate wrapper: the R1 idle grace and the broker_unavailable re-check
  counted only their sleeps against TIMEOUT, so python/herdr probe time plus
  fresh disk heartbeats let a job run far past --timeout (55 s for a 20 s
  budget in the new test). Both loops now end at START_TIME + TIMEOUT read
  off the clock on every pass; heartbeats extend only the idle window, and
  the herdr probe is cut off at the deadline (_within_deadline)
- document the window policy d5a191c actually implements: an agent with no
  evidence is closed ~2 x IDLE_TIMEOUT after its last wire event (Issue #6
  promised IDLE_TIMEOUT + 5 s) and heartbeats may be ~2 x IDLE_TIMEOUT apart;
  correct the claim that publish_event records a heartbeat on disk before
  the wire publish - it records right after the publish attempt
- shim ls: every failed `agent list` other than a structured
  server_not_running now exits nonzero with a "state unknown" diagnostic
  instead of reaching the list parser, where an error document on stdout
  read as an empty list and let drift A terminate live rows
- tests: stdout variants for down and unrelated errors, reconcile keeping
  herdr_confirmed False on an unrelated failure, and a wall-clock upper bound
  under continuous heartbeats and slow probes; the three new checks fail on
  d5a191c
- VERSIONS.md v4.4.2: items 3 and 6 and the test count (19)

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGpJXhSZasJLU38ksDghe
2026-09-13 01:53:57 +09:00

443 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""Issue #14 — a prompt that herdr accepted must not be reported as undelivered.
`herdr agent prompt --wait --until working --timeout 5000` answers `timeout` (or
`agent_prompt_stalled`) when the input was sent but no `working` state was
observed afterwards. Agents whose agent_status never leaves `idle` mid-turn
(codex under herdr 0.9.0) therefore had every delivered prompt closed as
`agent_blocked`, and their own `started` / `completed` events, published seconds
later, were refused as late. send_keys_safe now confirms an accepted submission
through the pane, and the delegate wrapper additionally honours the registry's
own acknowledgement before finalising a job as an inject failure.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import time
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SKILLS = REPO_ROOT / ".agents" / "skills"
LIB_SH = SKILLS / "lib.sh"
WRAPPER = SKILLS / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
REGISTRY_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
TIMEOUT_JSON = '{"error":{"code":"timeout","message":"timed out waiting for agent status"},"id":"cli:agent:prompt"}'
IDLE_JSON = '{"result":{"agent":{"agent_status":"idle"}}}'
def _base_env(cwd: Path, extra=None) -> dict:
env = dict(os.environ)
env["DELEGATE_JOB_PYTHON"] = sys.executable
env["AGENT_PYTHON_BIN"] = sys.executable
env["WORKSPACE_ROOT"] = str(cwd)
env["HOME"] = str(cwd)
if extra:
env.update({k: str(v) for k, v in extra.items()})
return env
def _run_lib_sh(snippet: str, cwd: Path, env=None, timeout=60) -> subprocess.CompletedProcess:
return subprocess.run(
["bash", "-c", f'source "{LIB_SH}" >/dev/null 2>&1; {snippet}'],
capture_output=True, text=True, cwd=str(cwd), env=_base_env(cwd, env), timeout=timeout,
)
def _codex_like_mock(pre_pane: str, post_pane: str, prompt_code: str = "timeout") -> str:
"""A bash `_sks_herdr` stand-in replaying what herdr 0.9.0 shows for codex:
the RPC returns `timeout`, agent_status stays `idle`, and only the pane
reveals what happened. Captures switch from PRE to POST once the prompt
has been submitted. Any paste-buffer use is a contract violation."""
return f'''
_pane_quiescent() {{ return 0; }}
_pane_dialog_open() {{ return 1; }}
FLAG="$WORKSPACE_ROOT/.prompt-submitted"
rm -f "$FLAG"
_sks_herdr() {{
case "${{1:-}}" in
agent)
case "${{2:-}}" in
prompt) touch "$FLAG"; echo '{{"error":{{"code":"{prompt_code}","message":"x"}}}}' >&2; return 1 ;;
get) echo '{IDLE_JSON}'; return 0 ;;
esac ;;
capture-pane)
if [ -e "$FLAG" ]; then printf '%s\\n' "{post_pane}"; else printf '%s\\n' "{pre_pane}"; fi
return 0 ;;
paste-buffer|set-buffer|delete-buffer|send-keys)
echo "PASTE_PATH_USED: $*" >&2; return 0 ;;
esac
return 0
}}
'''
CODEX_IDLE = "identified.\\n› Ask Codex to do anything"
CODEX_WORKING = "Working (0s • esc to interrupt)\\n› Ask Codex to do anything"
CODEX_ANSWERED = "• OK\\n› Ask Codex to do anything"
def _assert_single_injection(res: subprocess.CompletedProcess):
assert "PASTE_PATH_USED" not in res.stderr, f"must not fall back to paste-buffer: {res.stderr}"
# ---------------------------------------------------------------------------
# lib.sh: accepted submission confirmed through the pane
# ---------------------------------------------------------------------------
def test_accepted_timeout_confirmed_by_busy_token(tmp_path):
script = _codex_like_mock(CODEX_IDLE, CODEX_WORKING) + \
'send_keys_safe "reviewer-codex-01" "review the triage report" "job_14a"'
res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "5"})
assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
_assert_single_injection(res)
assert "delivery unknown" not in res.stderr
def test_accepted_timeout_confirmed_by_pane_change_after_fast_turn(tmp_path):
"""A turn that already finished before the first poll shows no busy token,
only the answer; the pane moving on from its pre-submit snapshot is enough."""
script = _codex_like_mock(CODEX_IDLE, CODEX_ANSWERED) + \
'send_keys_safe "reviewer-codex-01" "Reply with OK" "job_14b"'
res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "5"})
assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
_assert_single_injection(res)
def test_accepted_timeout_with_no_signal_still_fails_closed(tmp_path):
"""Nothing moves, nothing works: the grace window expires into rc=4 exactly
as before, without ever re-submitting."""
script = _codex_like_mock(CODEX_IDLE, CODEX_IDLE) + \
'send_keys_safe "reviewer-codex-01" "silent prompt" "job_14c"'
started = time.time()
res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "2"})
elapsed = time.time() - started
assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
assert "delivery unknown/failed" in res.stderr
_assert_single_injection(res)
assert 2 <= elapsed < 20, f"grace window not honoured: {elapsed:.1f}s"
def test_pre_existing_busy_token_is_not_taken_as_delivery(tmp_path):
"""A busy token that was already on screen before the prompt is a leftover
of the previous turn, not evidence for this one."""
script = _codex_like_mock(CODEX_WORKING, CODEX_WORKING) + \
'send_keys_safe "reviewer-codex-01" "prompt into a busy pane" "job_14d"'
res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "2"})
assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
_assert_single_injection(res)
def test_transport_error_is_not_treated_as_accepted(tmp_path):
"""Only `timeout` / `agent_prompt_stalled` mean the input was sent. A
transport-level failure keeps the single 2 s probe and rc=4, even if the
pane happens to change meanwhile (T2-13 contract, pinned with pane noise)."""
script = _codex_like_mock(CODEX_IDLE, CODEX_WORKING, prompt_code="transport_error") + \
'send_keys_safe "reviewer-codex-01" "prompt" "job_14e"'
started = time.time()
res = _run_lib_sh(script, tmp_path, env={"SKS_DELIVERY_GRACE_S": "30"})
elapsed = time.time() - started
assert res.returncode == 4, f"rc={res.returncode}\n{res.stderr}"
_assert_single_injection(res)
assert elapsed < 15, f"no grace poll may run for a non-accepted code: {elapsed:.1f}s"
def test_mock_herdr_timeout_with_idle_status_is_confirmed(tmp_path, mock_herdr):
"""Same scenario through the shared mock herdr: `timeout` from agent prompt,
agent_status pinned to idle, the pane gaining the busy token after
submission. Exactly one prompt call, no paste-buffer fallback."""
state_file = Path(mock_herdr)
state = json.loads(state_file.read_text())
state["agents"]["agt_codex"] = {
"agent": "codex",
"agent_status": "idle",
"prompt_wait_result": "timeout",
"buffer": "Ready",
}
state_file.write_text(json.dumps(state, indent=2))
res = _run_lib_sh(
'send_keys_safe "agt_codex" "review please" "job_14f"',
tmp_path, env={"SKS_IDLE_HANDSHAKE": "0", "SKS_DELIVERY_GRACE_S": "5"},
)
assert res.returncode == 0, f"rc={res.returncode}\n{res.stderr}"
final_state = json.loads(state_file.read_text())
calls = final_state.get("calls", [])
prompt_calls = [c for c in calls if len(c) >= 2 and c[0] == "agent" and c[1] == "prompt"]
paste_calls = [c for c in calls if len(c) >= 2 and (c[0] in ("paste-buffer", "set-buffer") or c[1] in ("paste-buffer", "set-buffer"))]
assert len(prompt_calls) == 1, f"prompt must be submitted exactly once, got {len(prompt_calls)}"
assert not paste_calls, f"no paste-buffer fallback allowed: {paste_calls}"
# ---------------------------------------------------------------------------
# wrapper: the registry's own acknowledgement beats an unconfirmed inject
# ---------------------------------------------------------------------------
def test_wrapper_honours_registry_ack_over_unconfirmed_inject(tmp_path, mock_herdr):
"""The RPC fails outright, but the agent has already published `started`
(status running) by the time the wrapper decides: the job must not be
closed as agent_blocked. With the broker unreachable the wrapper then
follows its normal broker_unavailable path (rc=3), never the inject path."""
reg_dir = tmp_path / "jobs"
reg_dir.mkdir(parents=True, exist_ok=True)
state_file = Path(mock_herdr)
state = json.loads(state_file.read_text())
state["agents"]["ack-agent"] = {
"agent": "codex",
"agent_status": "idle",
"prompt_wait_result": "transport_error",
}
state["prompt_wait_result"] = "transport_error"
state_file.write_text(json.dumps(state, indent=2))
env = _base_env(tmp_path, {
"MQTT_HOST": "127.0.0.1", "MQTT_PORT": "65534",
"SKS_IDLE_TIMEOUT_MS": "500", "MAM_INJECT_ACK_GRACE_S": "8",
})
proc = subprocess.Popen(
[
"bash", str(WRAPPER), "submit",
"--agent", "codex",
"--agent-session", "herdr:ack-agent",
"--prompt", "issue 14 ack",
"--registry-dir", str(reg_dir),
"--workdir", str(tmp_path),
"--timeout", "6",
"--idle-timeout", "2",
],
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
)
jid = ""
for _ in range(100):
job_files = list(reg_dir.glob("*.json"))
if job_files:
jid = job_files[0].stem
break
time.sleep(0.1)
assert jid, "job was not registered"
# The agent acknowledges the job out of band, as codex does via publish_event.
ack = subprocess.run(
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
"update", "--job", jid, "--status", "running"],
capture_output=True, text=True, cwd=str(tmp_path), env=env,
)
assert ack.returncode == 0, ack.stderr
stdout, stderr = proc.communicate(timeout=90)
assert "treating the prompt as delivered and continuing" in stderr, f"ack guard did not fire:\n{stderr}\n{stdout}"
assert "프롬프트 주입 실패" not in stderr
assert proc.returncode == 3, f"expected the broker_unavailable path (rc=3), got {proc.returncode}\n{stderr}"
events_log = reg_dir / f"{jid}.events.log"
if not events_log.exists():
events_log = reg_dir / jid / "events.log"
if events_log.exists():
assert "agent_blocked" not in events_log.read_text(), events_log.read_text()
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
assert job_data.get("status") != "pending"
# ---------------------------------------------------------------------------
# wrapper: idle-timeout grace (R1) must see a working codex through the pane
# ---------------------------------------------------------------------------
def test_wrapper_idle_timeout_grace_accepts_pane_busy_token(tmp_path, mock_herdr):
"""herdr keeps reporting `idle` for a codex-like agent that is visibly
working (busy token in the pane). When the subscriber idles out, the
wrapper must keep waiting for the on-disk result instead of closing the
job as deadline_timeout mid-turn; the agent then completes out of band."""
reg_dir = tmp_path / "jobs"
reg_dir.mkdir(parents=True, exist_ok=True)
state_file = Path(mock_herdr)
state = json.loads(state_file.read_text())
# `timeout` from agent prompt keeps agent_status idle; the mock appends the
# busy token to the pane on submission, exactly like a codex turn.
state["agents"]["slow-codex"] = {
"agent": "codex",
"agent_status": "idle",
"prompt_wait_result": "timeout",
"buffer": "Ready",
}
state_file.write_text(json.dumps(state, indent=2))
env = _base_env(tmp_path, {"SKS_IDLE_TIMEOUT_MS": "500", "SKS_DELIVERY_GRACE_S": "5"})
proc = subprocess.Popen(
[
"bash", str(WRAPPER), "submit",
"--agent", "codex",
"--agent-session", "herdr:slow-codex",
"--prompt", "issue 14 idle grace",
"--registry-dir", str(reg_dir),
"--workdir", str(tmp_path),
"--timeout", "40",
"--idle-timeout", "3",
],
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
)
jid = ""
for _ in range(100):
job_files = list(reg_dir.glob("*.json"))
if job_files:
jid = job_files[0].stem
break
time.sleep(0.1)
assert jid, "job was not registered"
# Well past the 3 s idle timeout, the agent finishes.
time.sleep(14)
done = subprocess.run(
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
"update", "--job", jid, "--status", "completed"],
capture_output=True, text=True, cwd=str(tmp_path), env=env,
)
assert done.returncode == 0, done.stderr
stdout, stderr = proc.communicate(timeout=90)
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
assert proc.returncode == 0, f"expected rc=0 after the working grace, got {proc.returncode}\n{stdout}\n{stderr}"
# ---------------------------------------------------------------------------
# wrapper: idle-timeout grace (R1) must honour on-disk progress heartbeats
# ---------------------------------------------------------------------------
PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
def test_wrapper_idle_timeout_grace_accepts_disk_heartbeat(tmp_path):
"""No herdr evidence at all (a human agent, or an agent running its tests
in a background shell): once the subscriber has idled out, the wrapper
must keep the job open as long as fresh `progress` heartbeats keep landing
in events.log, and pick up the eventual completion. Each heartbeat here
arrives after the previous idle window would have expired."""
reg_dir = tmp_path / "jobs"
reg_dir.mkdir(parents=True, exist_ok=True)
env = _base_env(tmp_path)
proc = subprocess.Popen(
[
"bash", str(WRAPPER), "submit",
"--agent", "human",
"--prompt", "issue 14 disk heartbeat",
"--registry-dir", str(reg_dir),
"--workdir", str(tmp_path),
"--timeout", "60",
"--idle-timeout", "6",
],
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
)
jid = ""
for _ in range(100):
job_files = list(reg_dir.glob("*.json"))
if job_files:
jid = job_files[0].stem
break
time.sleep(0.1)
assert jid, "job was not registered"
def heartbeat(detail: str):
subprocess.run(
[sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir),
"--job", jid, "--event", "progress", "--detail", detail],
capture_output=True, text=True, cwd=str(tmp_path), env=env,
)
# Subscriber idles out at ~6 s (plus broker connect); heartbeats then keep
# arriving 2 s apart, well inside the 6 s window each one re-opens (the
# wrapper's poll iteration costs ~2 s of herdr RPCs, so the ratio matters).
time.sleep(9)
for i in range(5):
heartbeat(f"still working {i}")
time.sleep(2)
done = subprocess.run(
[sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
"update", "--job", jid, "--status", "completed"],
capture_output=True, text=True, cwd=str(tmp_path), env=env,
)
assert done.returncode == 0, done.stderr
stdout, stderr = proc.communicate(timeout=90)
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
assert job_data.get("status") == "completed", f"status={job_data.get('status')}\n{stdout}\n{stderr}"
assert proc.returncode == 0, f"expected rc=0 while heartbeats kept landing, got {proc.returncode}\n{stdout}\n{stderr}"
def test_wrapper_idle_grace_never_outlives_the_wall_clock(tmp_path):
"""Heartbeats may extend the idle window, never the overall --timeout.
Fresh heartbeats land in events.log every second for the whole run, and
every python probe the wrapper makes is slowed down (herdr-RPC-like
latency). Counting sleeps instead of reading the clock let the grace loop
run about twice as long as TIMEOUT; the job must instead close as
deadline_timeout within TIMEOUT plus a small close/publish margin."""
import threading
timeout_s, idle_s = 20, 4
reg_dir = tmp_path / "jobs"
reg_dir.mkdir(parents=True, exist_ok=True)
slow_py = tmp_path / "slow-python"
slow_py.write_text(f'#!/bin/sh\nsleep 0.7\nexec "{sys.executable}" "$@"\n')
slow_py.chmod(0o755)
env = _base_env(tmp_path, {"DELEGATE_JOB_PYTHON": slow_py})
started = time.time()
proc = subprocess.Popen(
[
"bash", str(WRAPPER), "submit",
"--agent", "human",
"--prompt", "issue 14 wall clock",
"--registry-dir", str(reg_dir),
"--workdir", str(tmp_path),
"--timeout", str(timeout_s),
"--idle-timeout", str(idle_s),
],
cwd=str(tmp_path), env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
)
jid = ""
for _ in range(200):
job_files = list(reg_dir.glob("*.json"))
if job_files:
jid = job_files[0].stem
break
time.sleep(0.1)
assert jid, "job was not registered"
events_log = reg_dir / f"{jid}.events.log"
stop = threading.Event()
def keep_heartbeating():
# The same line shape publish_event appends; only its timestamp matters
# to _recent_disk_heartbeat.
seq = 10_000
while not stop.is_set():
seq += 1
line = {"schema_version": 1, "seq": seq, "job_id": jid, "event": "progress",
"timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"detail": "still working", "data": {}}
with open(events_log, "a") as fh:
fh.write(json.dumps(line) + "\n")
stop.wait(1)
beat = threading.Thread(target=keep_heartbeating, daemon=True)
beat.start()
try:
stdout, stderr = proc.communicate(timeout=120)
finally:
stop.set()
elapsed = time.time() - started
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
assert proc.returncode == 2, f"expected the deadline_timeout path (rc=2), got {proc.returncode}\n{stderr}"
assert job_data.get("status") == "error", f"status={job_data.get('status')}\n{stderr}"
# The heartbeats were honoured: the job was not closed after the first idle
# window, it stayed open up to the wall clock ...
assert elapsed >= timeout_s - 1, f"closed before the wall clock: {elapsed:.1f}s\n{stderr}"
# ... and not a moment past it beyond closing and publishing the error.
assert elapsed <= timeout_s + 12, f"idle grace outlived --timeout {timeout_s}s: {elapsed:.1f}s\n{stderr}"