#!/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 "acknowledged it; 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"