GodopuandClaude Fable 5.1
497633f64a
- Drop per-session agent runtime markers (CLAUDECODE, CLAUDE_CODE_CHILD_SESSION, CLAUDE_CODE_SESSION_ID, Claude Code messaging/bridge/pid markers, and the agy/hermes/grok/opencode session ids) from the environment lib.sh's shim hands to a bootstrapped `herdr --session <name> server`; user configuration such as CLAUDE_PROJECT_DIR still inherits - Add MAM_NO_AUTOSTART=1 so query-only callers never bootstrap a server as a side effect; status.sh and reconcile.sh export it, and the shim's `ls` reports a down server as "no server running" so reconcile keeps herdr_confirmed instead of freezing drift detection for every server - send_keys_safe: treat `timeout` / `agent_prompt_stalled` from `agent prompt --wait` as an accepted submission and confirm delivery for SKS_DELIVERY_GRACE_S (default 15 s) via agent_status working, a busy token the pane did not show before, or the pane moving on from its pre-submit snapshot; no signal at all still fails closed (rc 4). Codex under herdr 0.9.0 never reports `working`, so every delivered prompt was closed as agent_blocked and its later started/completed events refused - delegate wrapper: before closing an unconfirmed inject as agent_blocked, honour the registry's own acknowledgement for MAM_INJECT_ACK_GRACE_S (default 10 s) - Switch T1-7's failure trigger to transport_error (timeout now means accepted); add test_issue9_server_env.py (5 cases) and test_issue14_delivery_ack.py (7) - Verified live: a freshly bootstrapped multi-agent-mux server carries no markers and the claude panes save transcripts again; codex job d04ebcb5 had its started/completed accepted with no agent_blocked Closes #9 Closes #14 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VCgu9Fa9iT7TbJUSzdq4Ff
1225 lines
50 KiB
Python
1225 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""Issue #6 Phase 1 & Phase 3 — Fail-Closed & Fail-Fast regression test suite.
|
|
|
|
Tests T1-1 through T1-13 and T3-1 through T3-3 according to
|
|
docs/contracts/issue-6/implementation_plan.md (Rev.2).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
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"
|
|
REGISTRY_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "registry.py"
|
|
PUBLISH_PY = SKILLS / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py"
|
|
|
|
|
|
def _run_wrapper(args, env=None, cwd=None, timeout=60):
|
|
run_env = dict(os.environ)
|
|
run_env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
run_env["AGENT_PYTHON_BIN"] = sys.executable
|
|
if cwd:
|
|
run_env["WORKSPACE_ROOT"] = str(cwd)
|
|
run_env["HOME"] = str(cwd)
|
|
if env:
|
|
run_env.update({k: str(v) for k, v in env.items()})
|
|
return subprocess.run(
|
|
[str(WRAPPER)] + list(args),
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=str(cwd or REPO_ROOT),
|
|
env=run_env,
|
|
timeout=timeout,
|
|
)
|
|
|
|
|
|
def _run_registry(subcmd, args, reg_dir, cwd=None):
|
|
py_bin = sys.executable
|
|
cmd = [py_bin, str(REGISTRY_PY), "--registry-dir", str(reg_dir), subcmd] + list(args)
|
|
return subprocess.run(cmd, capture_output=True, text=True, cwd=str(cwd or REPO_ROOT))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-1: direct, 이벤트 없음 -> rc=2, 레지스트리 error, events.log 1건, subscriber pid 소멸
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_1_direct_no_event_timeout_fail_closed(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "human",
|
|
"--prompt", "test T1-1 prompt",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "6",
|
|
"--idle-timeout", "2",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 2, f"Expected rc=2, got {res.returncode}. Stderr: {res.stderr}"
|
|
|
|
# Extract job ID
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "error", f"Expected error status, got {job_data.get('status')}"
|
|
|
|
# Check events.log
|
|
jid = job_data["job_id"]
|
|
events_log = reg_dir / jid / "events.log"
|
|
if events_log.exists():
|
|
lines = [json.loads(l) for l in events_log.read_text().splitlines() if l.strip()]
|
|
error_events = [e for e in lines if e.get("event") == "error"]
|
|
assert len(error_events) >= 1
|
|
|
|
# Check subscriber log indicates timeout
|
|
sub_log = reg_dir / f"{jid}.subscriber.out"
|
|
assert sub_log.exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-2: loop, 워커 phase 이벤트 없음 -> rc=2, error, retained 발행 없음
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_2_loop_worker_no_event_fail_closed(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--type", "loop",
|
|
"--agent", "human",
|
|
"--reviewer", "human",
|
|
"--prompt", "test T1-2 prompt",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "6",
|
|
"--idle-timeout", "2",
|
|
"--max-iterations", "1",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 2, f"Expected rc=2, got {res.returncode}. Stderr: {res.stderr}"
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "error"
|
|
|
|
# Verify no retained error was published for loop
|
|
jid = job_data["job_id"]
|
|
events_log = reg_dir / jid / "events.log"
|
|
if events_log.exists():
|
|
lines = [json.loads(l) for l in events_log.read_text().splitlines() if l.strip()]
|
|
wrapper_errors = [e for e in lines if e.get("source") == "delegate_wrapper"]
|
|
assert len(wrapper_errors) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-3: 구독자 타임아웃 직후 publish_event completed 경쟁
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_3_race_completed_beats_timeout_close(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 1. Register a job
|
|
reg_res = _run_registry("register", ["--prompt", "race", "--agent", "claude"], reg_dir)
|
|
assert reg_res.returncode == 0
|
|
jid = reg_res.stdout.strip().split()[-1]
|
|
|
|
# 2. Mark completed
|
|
comp_res = _run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
assert comp_res.returncode == 0
|
|
|
|
# 3. Simulate subscriber timeout close attempt
|
|
close_res = _run_registry("close", ["--job", jid, "--request", "deadline_timeout"], reg_dir)
|
|
assert close_res.returncode == 0 # 0 on terminal refusal / idempotent
|
|
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "completed", "Status must remain completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-4: MQTT_HOST 불가 + 디스크 비종료 -> rc=3, 데드라인 재관측 후 error
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_4_broker_unavailable_fail_closed(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "human",
|
|
"--prompt", "test T1-4 prompt",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "4",
|
|
"--idle-timeout", "2",
|
|
],
|
|
env={"MQTT_HOST": "127.0.0.1", "MQTT_PORT": "65534"},
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 3, f"Expected rc=3, got {res.returncode}. Stderr: {res.stderr}"
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "error"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-5: rc=3 + 재관측 중 디스크 completed 출현 -> rc=0, completed
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_5_broker_unavailable_reobserve_completed(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
run_env = dict(os.environ, MQTT_HOST="127.0.0.1", MQTT_PORT="65534")
|
|
run_env["DELEGATE_JOB_PYTHON"] = sys.executable
|
|
run_env["AGENT_PYTHON_BIN"] = sys.executable
|
|
run_env["WORKSPACE_ROOT"] = str(tmp_path)
|
|
run_env["HOME"] = str(tmp_path)
|
|
|
|
proc = subprocess.Popen(
|
|
[
|
|
str(WRAPPER),
|
|
"submit",
|
|
"--agent", "human",
|
|
"--prompt", "test T1-5 prompt",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "6",
|
|
"--idle-timeout", "2",
|
|
],
|
|
env=run_env,
|
|
cwd=str(tmp_path),
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
|
|
# Wait for job file to appear
|
|
jid = ""
|
|
for _ in range(50):
|
|
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"
|
|
time.sleep(1.0)
|
|
|
|
# Simulate out-of-band completion via registry update
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
|
|
stdout, stderr = proc.communicate(timeout=60)
|
|
assert proc.returncode == 0, f"Expected rc=0, got {proc.returncode}. Stderr: {stderr}"
|
|
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-6: close --request deadline_timeout을 completed 레코드에 적용 -> rc=0 불변
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_6_close_deadline_timeout_on_completed_record(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
reg_res = _run_registry("register", ["--prompt", "t1-6", "--agent", "claude"], reg_dir)
|
|
jid = reg_res.stdout.strip().split()[-1]
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
|
|
close_res = _run_registry("close", ["--job", jid, "--request", "deadline_timeout"], reg_dir)
|
|
assert close_res.returncode == 0
|
|
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-7: 커스텀 --registry-dir + send_keys_safe 실패 -> rc=1, 해당 레지스트리 error
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_7_custom_registry_dir_inject_failure(tmp_path, mock_herdr):
|
|
reg_dir = tmp_path / "custom_jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Set up mock herdr with an agent, but prompt will fail
|
|
state_file = Path(mock_herdr)
|
|
state = json.loads(state_file.read_text())
|
|
# A transport-level RPC failure is a definite inject failure. (`timeout` is
|
|
# not: since Issue #14 it means herdr accepted the submission and only its
|
|
# status watch timed out, which send_keys_safe then confirms via the pane.)
|
|
state["agents"]["fail-agent"] = {
|
|
"agent": "claude",
|
|
"agent_status": "idle",
|
|
"prompt_wait_result": "transport_error", # will fail prompt
|
|
}
|
|
state["prompt_wait_result"] = "transport_error"
|
|
state_file.write_text(json.dumps(state, indent=2))
|
|
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "claude",
|
|
"--agent-session", "herdr:fail-agent",
|
|
"--prompt", "test T1-7",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "15",
|
|
"--idle-timeout", "10",
|
|
],
|
|
env={"SKS_BUSY_TIMEOUT": "1", "SKS_IDLE_TIMEOUT_MS": "500", "MAM_INJECT_ACK_GRACE_S": "1"},
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 1, f"Expected rc=1, got {res.returncode}. Stderr: {res.stderr}"
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "error"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-8: MAM_DELEGATE_FAIL_CLOSED=0 -> rc=0, pending (구 동작 재현)
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_8_rollback_drill_legacy_pending(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "human",
|
|
"--prompt", "test T1-8 prompt",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "4",
|
|
"--idle-timeout", "2",
|
|
],
|
|
env={"MAM_DELEGATE_FAIL_CLOSED": "0"},
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
assert res.returncode == 0, f"Expected rc=0 under rollback, got {res.returncode}"
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "pending", f"Expected pending under legacy, got {job_data.get('status')}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-9: dry-run direct -> 구독자 프로세스 미기동
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_9_dry_run_direct_no_subscriber(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "human",
|
|
"--prompt", "test T1-9 dry-run",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--dry-run",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=30,
|
|
)
|
|
assert res.returncode == 0
|
|
assert "[dry-run]" in res.stdout
|
|
sub_logs = list(reg_dir.glob("*.subscriber.out"))
|
|
assert len(sub_logs) == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-10: loop: 워커 completed 후 리뷰어 update --status pending --iteration 1 --attempt-id 2
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_10_attempt_id_advances_same_iteration(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
reg_res = _run_registry("register", ["--prompt", "t1-10", "--agent", "claude"], reg_dir)
|
|
jid = reg_res.stdout.strip().split()[-1]
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
|
|
# Now update with attempt_id=2 on iteration 1
|
|
upd_res = _run_registry("update", ["--job", jid, "--status", "pending", "--iteration", "1", "--attempt-id", "2"], reg_dir)
|
|
assert upd_res.returncode == 0, f"Expected rc=0, got {upd_res.returncode}. Stderr: {upd_res.stderr}"
|
|
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "pending"
|
|
assert job_data.get("attempt_id") == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-11: 같은 --attempt-id 2 재요청(리플레이) -> rc=4 거부
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_11_replay_same_attempt_id_rejected(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
reg_res = _run_registry("register", ["--prompt", "t1-11", "--agent", "claude"], reg_dir)
|
|
jid = reg_res.stdout.strip().split()[-1]
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
|
|
_run_registry("update", ["--job", jid, "--status", "pending", "--iteration", "1", "--attempt-id", "2"], reg_dir)
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
|
|
# Replay attempt 2 -> must be rejected rc=4
|
|
replay_res = _run_registry("update", ["--job", jid, "--status", "pending", "--iteration", "1", "--attempt-id", "2"], reg_dir)
|
|
assert replay_res.returncode == 4, f"Expected rc=4 for replayed attempt_id, got {replay_res.returncode}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-12: --attempt-id 없이 iteration 2 (F-A 경로) -> rc=0
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_12_legacy_iteration_increment_without_attempt_id(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
reg_res = _run_registry("register", ["--prompt", "t1-12", "--agent", "claude"], reg_dir)
|
|
jid = reg_res.stdout.strip().split()[-1]
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
|
|
# Iteration 2 without attempt_id -> advances past 1
|
|
upd_res = _run_registry("update", ["--job", jid, "--status", "pending", "--iteration", "2"], reg_dir)
|
|
assert upd_res.returncode == 0, f"Expected rc=0, got {upd_res.returncode}"
|
|
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "pending"
|
|
assert job_data.get("iteration") == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-13: close --request cancelled --detail session_busy 전이 규칙 고정
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_13_close_cancelled_transition_rules(tmp_path):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 1. pending -> cancelled rc=0
|
|
reg_res = _run_registry("register", ["--prompt", "t1-13", "--agent", "claude"], reg_dir)
|
|
jid = reg_res.stdout.strip().split()[-1]
|
|
|
|
close_res = _run_registry("close", ["--job", jid, "--request", "cancelled", "--detail", "session_busy"], reg_dir)
|
|
assert close_res.returncode == 0
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "cancelled"
|
|
assert job_data.get("detail") == "session_busy"
|
|
|
|
# 2. cancelled -> redispatch (iteration 2) -> pending rc=0
|
|
redisp_res = _run_registry("update", ["--job", jid, "--status", "pending", "--iteration", "2"], reg_dir)
|
|
assert redisp_res.returncode == 0
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "pending"
|
|
|
|
# 3. completed -> cancelled -> rc=0, 상태 불변
|
|
_run_registry("update", ["--job", jid, "--status", "completed"], reg_dir)
|
|
close_res2 = _run_registry("close", ["--job", jid, "--request", "cancelled", "--detail", "session_busy"], reg_dir)
|
|
assert close_res2.returncode == 0
|
|
job_data = json.loads((reg_dir / f"{jid}.json").read_text())
|
|
assert job_data.get("status") == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T3-1: LoopRunner.delegate_job rc!=0 + registered job: 출력 -> id 반환, wait_for_job <= 2s False
|
|
# ---------------------------------------------------------------------------
|
|
def test_t3_1_runner_delegate_job_fail_fast(tmp_path, monkeypatch):
|
|
from lib_py.loop.runner import LoopRunner
|
|
|
|
def fake_run(*args, **kwargs):
|
|
res = MagicMock()
|
|
res.returncode = 2
|
|
res.stdout = "registered job: jid123\nsubscriber pid: 12345\n"
|
|
res.stderr = "timeout\n"
|
|
return res
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
|
runner = LoopRunner()
|
|
|
|
jid = runner.delegate_job("herdr:claude", "claude", "Worker", "test prompt")
|
|
assert jid == "jid123"
|
|
|
|
t0 = time.time()
|
|
res = runner.wait_for_job(jid, max_wait=2)
|
|
elapsed = time.time() - t0
|
|
assert res is False
|
|
assert elapsed <= 2.5
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T3-2: LoopRunner.delegate_job rc!=0 -> checkpoint에 id 기록
|
|
# ---------------------------------------------------------------------------
|
|
def test_t3_2_runner_delegate_job_records_checkpoint(tmp_path, monkeypatch):
|
|
from lib_py.loop.runner import LoopRunner
|
|
|
|
def fake_run(*args, **kwargs):
|
|
res = MagicMock()
|
|
res.returncode = 2
|
|
res.stdout = "registered job: jid_chk123\n"
|
|
res.stderr = "error\n"
|
|
return res
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
|
runner = LoopRunner()
|
|
|
|
jid = runner.delegate_job("herdr:claude", "claude", "Worker", "test prompt")
|
|
assert jid == "jid_chk123"
|
|
|
|
checkpoint_file = tmp_path / ".mam" / "loop-checkpoint.json"
|
|
assert checkpoint_file.exists()
|
|
chk_data = json.loads(checkpoint_file.read_text())
|
|
assert "jid_chk123" in chk_data.get("created_jobs", [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T3-3: 래퍼 stub이 1회차 rc=5, 2회차 rc=0 -> 재위임 1회 후 2회차 id 반환
|
|
# ---------------------------------------------------------------------------
|
|
def test_t3_3_runner_delegate_job_session_busy_retry(tmp_path, monkeypatch):
|
|
from lib_py.loop.runner import LoopRunner
|
|
|
|
call_count = 0
|
|
|
|
def fake_run(*args, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
res = MagicMock()
|
|
if call_count == 1:
|
|
res.returncode = 5
|
|
res.stdout = "registered job: jid_busy_1\n"
|
|
res.stderr = "session busy\n"
|
|
else:
|
|
res.returncode = 0
|
|
res.stdout = "registered job: jid_busy_2\n"
|
|
res.stderr = ""
|
|
return res
|
|
|
|
monkeypatch.setattr(subprocess, "run", fake_run)
|
|
monkeypatch.setenv("WORKSPACE_ROOT", str(tmp_path))
|
|
monkeypatch.setenv("MAM_SESSION_BUSY_RETRY_S", "0")
|
|
|
|
runner = LoopRunner()
|
|
jid = runner.delegate_job("herdr:claude", "claude", "Worker", "test prompt")
|
|
assert call_count == 2
|
|
assert jid == "jid_busy_2"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-14: R1 working 중 subscriber 타임아웃 -> 디스크 completed 재관측 수용 (rc=0)
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_14_working_agent_reobserve_completed(tmp_path, mock_herdr):
|
|
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"]["agt_w1"] = {
|
|
"agent": "claude",
|
|
"agent_status": "idle",
|
|
}
|
|
state_file.write_text(json.dumps(state, indent=2))
|
|
|
|
import threading
|
|
|
|
stop_ev = threading.Event()
|
|
def watch_and_progress():
|
|
while not stop_ev.is_set():
|
|
try:
|
|
st = json.loads(state_file.read_text())
|
|
if any("agent" in c and "prompt" in c for c in st.get("calls", [])):
|
|
st["agents"]["agt_w1"]["agent_status"] = "working"
|
|
state_file.write_text(json.dumps(st, indent=2))
|
|
break
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.05)
|
|
# Wait for subscriber idle timeout (3s) so wrapper enters R1 working loop
|
|
deadline = time.time() + 10.0
|
|
while time.time() < deadline and not stop_ev.is_set():
|
|
sub_logs = list(reg_dir.glob("*.subscriber.out"))
|
|
if sub_logs:
|
|
try:
|
|
content = sub_logs[0].read_text(errors="ignore")
|
|
if "idle timeout" in content:
|
|
break
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.1)
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
if job_files:
|
|
jf = job_files[0]
|
|
try:
|
|
data = json.loads(jf.read_text())
|
|
data["status"] = "completed"
|
|
jf.write_text(json.dumps(data))
|
|
except Exception:
|
|
pass
|
|
|
|
t = threading.Thread(target=watch_and_progress)
|
|
t.daemon = True
|
|
t.start()
|
|
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "claude",
|
|
"--agent-session", "herdr:agt_w1",
|
|
"--prompt", "test T1-14",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "15",
|
|
"--idle-timeout", "3",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
stop_ev.set()
|
|
assert res.returncode == 0, f"Expected rc=0 on late completed, got {res.returncode}. Stderr: {res.stderr}"
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "completed"
|
|
final_st = json.loads(state_file.read_text())
|
|
agent_gets = [c for c in final_st.get("calls", []) if "agent" in c and "get" in c]
|
|
assert len(agent_gets) >= 1, "R1 loop must query agent get"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-15: R1 working 고정 + 미발행 -> 벽시계 --timeout 만료 후 rc=2 error
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_15_working_agent_reobserve_wallclock_timeout(tmp_path, mock_herdr):
|
|
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"]["agt_w2"] = {
|
|
"agent": "claude",
|
|
"agent_status": "idle",
|
|
}
|
|
state_file.write_text(json.dumps(state, indent=2))
|
|
|
|
import threading
|
|
|
|
stop_ev = threading.Event()
|
|
def watch_and_stay_working():
|
|
while not stop_ev.is_set():
|
|
try:
|
|
st = json.loads(state_file.read_text())
|
|
if any("agent" in c and "prompt" in c for c in st.get("calls", [])):
|
|
st["agents"]["agt_w2"]["agent_status"] = "working"
|
|
state_file.write_text(json.dumps(st, indent=2))
|
|
break
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.05)
|
|
|
|
t = threading.Thread(target=watch_and_stay_working)
|
|
t.daemon = True
|
|
t.start()
|
|
|
|
start_t = time.time()
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "claude",
|
|
"--agent-session", "herdr:agt_w2",
|
|
"--prompt", "test T1-15",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "5",
|
|
"--idle-timeout", "2",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
stop_ev.set()
|
|
elapsed = time.time() - start_t
|
|
assert res.returncode == 2, f"Expected rc=2 on wallclock timeout, got {res.returncode}"
|
|
assert elapsed >= 4.0, f"Must wait up to wallclock timeout, elapsed={elapsed}"
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "error"
|
|
final_st = json.loads(state_file.read_text())
|
|
agent_gets = [c for c in final_st.get("calls", []) if "agent" in c and "get" in c]
|
|
assert len(agent_gets) >= 2, "R1 loop must repeatedly query agent get during wallclock wait"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-16: R1 working 중 done 전환 + 미발행 -> 1초 유예 후 rc=2 error
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_16_working_agent_done_without_event(tmp_path, mock_herdr):
|
|
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"]["agt_w3"] = {
|
|
"agent": "claude",
|
|
"agent_status": "idle",
|
|
}
|
|
state_file.write_text(json.dumps(state, indent=2))
|
|
|
|
import threading
|
|
|
|
stop_ev = threading.Event()
|
|
def watch_and_finish():
|
|
while not stop_ev.is_set():
|
|
try:
|
|
st = json.loads(state_file.read_text())
|
|
if any("agent" in c and "prompt" in c for c in st.get("calls", [])):
|
|
st["agents"]["agt_w3"]["agent_status"] = "working"
|
|
state_file.write_text(json.dumps(st, indent=2))
|
|
break
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.05)
|
|
time.sleep(2.8)
|
|
try:
|
|
st = json.loads(state_file.read_text())
|
|
st["agents"]["agt_w3"]["agent_status"] = "done"
|
|
state_file.write_text(json.dumps(st, indent=2))
|
|
except Exception:
|
|
pass
|
|
|
|
t = threading.Thread(target=watch_and_finish)
|
|
t.daemon = True
|
|
t.start()
|
|
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "claude",
|
|
"--agent-session", "herdr:agt_w3",
|
|
"--prompt", "test T1-16",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "15",
|
|
"--idle-timeout", "2",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
stop_ev.set()
|
|
assert res.returncode == 2, f"Expected rc=2 on done without event, got {res.returncode}"
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
assert len(job_files) == 1
|
|
job_data = json.loads(job_files[0].read_text())
|
|
assert job_data.get("status") == "error"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-17: publish_event with mismatched --expect-dispatch-token exits rc=4 without wire publish
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_17_publish_event_token_mismatch_rc4_no_wire(tmp_path, monkeypatch):
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
from lib_py.delegation import registry, publish_event
|
|
|
|
job_id = registry.register_job(
|
|
job_id="test_t1_17_job",
|
|
prompt="prompt",
|
|
agent="claude",
|
|
agent_session="herdr:claude",
|
|
registry_dir=str(reg_dir),
|
|
)
|
|
rec = registry.load_job(job_id, str(reg_dir))
|
|
real_token = rec["dispatch_token"]
|
|
|
|
published_calls = []
|
|
def fake_publish_once(config, topic, body, retain):
|
|
published_calls.append((topic, body, retain))
|
|
|
|
monkeypatch.setattr(publish_event, "_publish_once", fake_publish_once)
|
|
|
|
rc = publish_event.main([
|
|
"--job", job_id,
|
|
"--registry-dir", str(reg_dir),
|
|
"--event", "completed",
|
|
"--detail", "test",
|
|
"--expect-dispatch-token", "wrong-token-12345",
|
|
])
|
|
assert rc == 4, f"Expected rc=4 on token mismatch, got {rc}"
|
|
assert len(published_calls) == 0, "No message must be sent over MQTT wire when token mismatches!"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-18: job_subscriber drops events with mismatched dispatch token
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_18_job_subscriber_drops_stale_dispatch_token(tmp_path):
|
|
from lib_py.delegation.job_subscriber import _Watcher
|
|
|
|
watcher = _Watcher(
|
|
expected_job_ids={"job_1"},
|
|
expected_tokens={"job_1": None},
|
|
expected_seqs={"job_1": 0},
|
|
expected_dispatch_tokens={"job_1": "token_active"},
|
|
expected_attempts={"job_1": 2},
|
|
)
|
|
|
|
class FakeMsg:
|
|
topic = "python/mqtt/jobs/job_1/events"
|
|
def __init__(self, payload_dict):
|
|
self.payload = json.dumps(payload_dict).encode("utf-8")
|
|
|
|
stale_payload = {
|
|
"schema_version": 1,
|
|
"job_id": "job_1",
|
|
"seq": 1,
|
|
"event": "error",
|
|
"data": {"dispatch_token": "token_old", "attempt_id": 1}
|
|
}
|
|
watcher.on_message(None, None, FakeMsg(stale_payload))
|
|
assert watcher.events.empty(), "Watcher must drop event with mismatched dispatch token!"
|
|
|
|
valid_payload = {
|
|
"schema_version": 1,
|
|
"job_id": "job_1",
|
|
"seq": 2,
|
|
"event": "completed",
|
|
"data": {"dispatch_token": "token_active", "attempt_id": 2}
|
|
}
|
|
watcher.on_message(None, None, FakeMsg(valid_payload))
|
|
assert not watcher.events.empty(), "Watcher must accept event with matching dispatch token!"
|
|
topic, ev = watcher.events.get()
|
|
assert ev["event"] == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-19: register_job initializes unique dispatch_token and attempt_id
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_19_fresh_direct_job_has_dispatch_token(tmp_path):
|
|
from lib_py.delegation import registry
|
|
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
jid1 = registry.register_job("job_d1", "prompt 1", "claude", "herdr:claude", registry_dir=str(reg_dir))
|
|
jid2 = registry.register_job("job_d2", "prompt 2", "claude", "herdr:claude", registry_dir=str(reg_dir))
|
|
|
|
rec1 = registry.load_job(jid1, str(reg_dir))
|
|
rec2 = registry.load_job(jid2, str(reg_dir))
|
|
|
|
assert rec1.get("dispatch_token") is not None
|
|
assert rec2.get("dispatch_token") is not None
|
|
assert rec1["dispatch_token"] != rec2["dispatch_token"]
|
|
assert rec1.get("attempt_id") == 1
|
|
assert rec2.get("attempt_id") == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-20: loop mode worker late completion continues to reviewer phase
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_20_loop_worker_late_completed_continues_to_reviewer(tmp_path, mock_herdr):
|
|
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"]["agt_worker"] = {
|
|
"agent": "claude",
|
|
"agent_status": "idle",
|
|
}
|
|
state["agents"]["agt_reviewer"] = {
|
|
"agent": "opencode",
|
|
"agent_status": "idle",
|
|
}
|
|
state_file.write_text(json.dumps(state, indent=2))
|
|
|
|
import threading
|
|
stop_ev = threading.Event()
|
|
def watch_worker():
|
|
while not stop_ev.is_set():
|
|
try:
|
|
st = json.loads(state_file.read_text())
|
|
if any("agent" in c and "prompt" in c for c in st.get("calls", [])):
|
|
st["agents"]["agt_worker"]["agent_status"] = "working"
|
|
state_file.write_text(json.dumps(st, indent=2))
|
|
break
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.05)
|
|
# Wait for worker subscriber idle timeout (3s)
|
|
deadline = time.time() + 25.0
|
|
while time.time() < deadline and not stop_ev.is_set():
|
|
sub_logs = list(reg_dir.glob("*.iter_1_Worker.subscriber.out"))
|
|
if sub_logs:
|
|
try:
|
|
content = sub_logs[0].read_text(errors="ignore")
|
|
if "idle timeout" in content:
|
|
break
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.1)
|
|
|
|
job_files = list(reg_dir.glob("*.json"))
|
|
if job_files:
|
|
jf = job_files[0]
|
|
try:
|
|
data = json.loads(jf.read_text())
|
|
data["status"] = "completed"
|
|
data["last_feedback"] = "worker finished task"
|
|
jf.write_text(json.dumps(data))
|
|
except Exception:
|
|
pass
|
|
|
|
t = threading.Thread(target=watch_worker)
|
|
t.daemon = True
|
|
t.start()
|
|
|
|
res = _run_wrapper(
|
|
[
|
|
"submit",
|
|
"--agent", "claude",
|
|
"--agent-session", "herdr:agt_worker",
|
|
"--reviewer", "opencode",
|
|
"--reviewer-session", "herdr:agt_reviewer",
|
|
"--prompt", "loop task",
|
|
"--type", "loop",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--timeout", "12",
|
|
"--idle-timeout", "2",
|
|
],
|
|
cwd=tmp_path,
|
|
timeout=60,
|
|
)
|
|
stop_ev.set()
|
|
assert "Worker completed during grace period. Continuing to reviewer." in res.stdout
|
|
final_st = json.loads(state_file.read_text())
|
|
rev_prompts = [c for c in final_st.get("calls", []) if "agent" in c and "prompt" in c and "agt_reviewer" in c]
|
|
assert len(rev_prompts) >= 1, "Reviewer must be dispatched after worker completes during grace period"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T3-4: wait_for_job 레코드 부재 및 실패 submit 5초 내 빠른 탈출
|
|
# ---------------------------------------------------------------------------
|
|
def test_t3_4_wait_for_job_absent_and_failed_submit_fast_fail(tmp_path):
|
|
from lib_py.loop.runner import LoopRunner
|
|
|
|
runner = LoopRunner()
|
|
runner.workspace_root = tmp_path
|
|
runner.mam_dir = tmp_path / ".mam"
|
|
(tmp_path / ".mam" / "jobs").mkdir(parents=True, exist_ok=True)
|
|
|
|
# 1) Non-existent job id with max_wait=30 -> aborts in <= 6.5s
|
|
start_t = time.time()
|
|
res1 = runner.wait_for_job("absent_job_xyz", max_wait=30)
|
|
elapsed1 = time.time() - start_t
|
|
assert res1 is False
|
|
assert elapsed1 <= 6.5, f"Absent job wait must abort in ~5s, took {elapsed1}s"
|
|
|
|
# 2) Failed submit job in failed_submit_jobs with nonterminal file -> aborts in <= 6.5s
|
|
failed_jid = "failed_submit_123"
|
|
runner.failed_submit_jobs.add(failed_jid)
|
|
jf = tmp_path / ".mam" / "jobs" / f"{failed_jid}.json"
|
|
jf.write_text(json.dumps({"job_id": failed_jid, "status": "running"}))
|
|
|
|
start_t = time.time()
|
|
res2 = runner.wait_for_job(failed_jid, max_wait=30)
|
|
elapsed2 = time.time() - start_t
|
|
assert res2 is False
|
|
assert elapsed2 <= 6.5, f"Failed submit job wait must abort in ~5s, took {elapsed2}s"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-21: Worker instructions bake dispatch_token & attempt_id; late attempt rejected
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_21_worker_instructions_bake_token_and_reject_stale_attempt(tmp_path):
|
|
import re, uuid
|
|
from lib_py.delegation import registry, mqtt_common, engine
|
|
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 1) Direct mode: register job and check instructions
|
|
jid = registry.register_job("job_p1", "test brief", "claude", "herdr:claude", registry_dir=str(reg_dir))
|
|
rec1 = registry.load_job(jid, str(reg_dir))
|
|
tok1 = rec1["dispatch_token"]
|
|
assert tok1, "Attempt 1 must have a dispatch token"
|
|
assert rec1.get("attempt_id") == 1
|
|
|
|
# Verify dry-run wrapper generates instructions containing baked token and attempt_id
|
|
res_dry = _run_wrapper([
|
|
"submit",
|
|
"--agent", "claude",
|
|
"--agent-session", "herdr:claude",
|
|
"--prompt", "test dry run prompt",
|
|
"--registry-dir", str(reg_dir),
|
|
"--workdir", str(tmp_path),
|
|
"--job", jid,
|
|
"--dry-run"
|
|
])
|
|
# Check that dry-run succeeded
|
|
assert res_dry.returncode == 0, f"dry-run failed: {res_dry.stderr}"
|
|
|
|
# 2) Complete or timeout attempt 1 and redispatch to Attempt 2
|
|
engine.transition(jid, "deadline_timeout", registry_dir=str(reg_dir), source="delegate_wrapper")
|
|
res_redisp = engine.transition(jid, "redispatch", registry_dir=str(reg_dir), source="operator",
|
|
fields={"iteration": 1, "attempt_id": 2})
|
|
assert res_redisp.committed, f"redispatch failed: {res_redisp.reason}"
|
|
rec2 = registry.load_job(jid, str(reg_dir))
|
|
assert rec2["status"] == "pending"
|
|
tok2 = rec2["dispatch_token"]
|
|
assert tok2 != tok1, "Redispatch must mint a new dispatch_token"
|
|
assert rec2["attempt_id"] == 2
|
|
|
|
# 3) Late worker from attempt 1 executes its saved completion command with tok1
|
|
pub_py = str(Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py")
|
|
late_cmd = [
|
|
sys.executable, pub_py,
|
|
"--registry-dir", str(reg_dir),
|
|
"--job", jid,
|
|
"--event", "completed",
|
|
"--detail", "late attempt 1 finish",
|
|
"--expect-dispatch-token", tok1,
|
|
"--attempt-id", "1"
|
|
]
|
|
late_res = subprocess.run(late_cmd, capture_output=True, text=True)
|
|
assert late_res.returncode == 4, f"Expected rc=4 for stale dispatch token, got {late_res.returncode}. Stderr: {late_res.stderr}"
|
|
|
|
# Disk status must remain pending (not overwritten by attempt 1)
|
|
rec_after = registry.load_job(jid, str(reg_dir))
|
|
assert rec_after["status"] == "pending", f"Job status must remain pending, got {rec_after['status']}"
|
|
|
|
# 4) Also verify that if publish_event is called with stale token in --data, it is rejected
|
|
stale_data_cmd = [
|
|
sys.executable, pub_py,
|
|
"--registry-dir", str(reg_dir),
|
|
"--job", jid,
|
|
"--event", "completed",
|
|
"--detail", "late attempt 1 finish via data",
|
|
"--data", json.dumps({"dispatch_token": tok1, "attempt_id": 1})
|
|
]
|
|
stale_data_res = subprocess.run(stale_data_cmd, capture_output=True, text=True)
|
|
assert stale_data_res.returncode == 4, f"Expected rc=4 for stale token in --data, got {stale_data_res.returncode}"
|
|
rec_after_data = registry.load_job(jid, str(reg_dir))
|
|
assert rec_after_data["status"] == "pending"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-22: job_subscriber disk fallback rejects stale dispatch token / attempt
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_22_subscriber_disk_fallback_rejects_stale_dispatch_token(tmp_path):
|
|
from lib_py.delegation import registry
|
|
from lib_py.delegation.job_subscriber import _check_disk_fallback
|
|
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
jid = registry.register_job("job_fb1", "prompt fb", "claude", "herdr:claude", registry_dir=str(reg_dir))
|
|
rec = registry.load_job(jid, str(reg_dir))
|
|
current_token = rec["dispatch_token"]
|
|
|
|
# Pretend disk record is completed with current_token
|
|
rec["status"] = "completed"
|
|
(reg_dir / f"{jid}.json").write_text(json.dumps(rec))
|
|
|
|
pending = {jid}
|
|
terminal = {}
|
|
|
|
# Case A: subscriber expecting a DIFFERENT dispatch token must ignore this completed status
|
|
expected_tokens = {jid: "some_other_newer_token"}
|
|
expected_attempts = {jid: 2}
|
|
_check_disk_fallback(pending, str(reg_dir), terminal, expected_tokens, expected_attempts)
|
|
assert jid in pending, "Pending job must NOT be removed when dispatch_token mismatches"
|
|
assert jid not in terminal, "Terminal dict must not record mismatched fallback status"
|
|
|
|
# Case B: subscriber expecting the MATCHING dispatch token accepts fallback
|
|
expected_matching = {jid: current_token}
|
|
expected_matching_att = {jid: 1}
|
|
_check_disk_fallback(pending, str(reg_dir), terminal, expected_matching, expected_matching_att)
|
|
assert jid not in pending, "Pending job must be resolved when dispatch_token matches"
|
|
assert terminal.get(jid) == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-23: DelegationService.publish_event enforces CAS token checks
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_23_service_publish_event_enforces_cas_token(tmp_path):
|
|
from lib_py.delegation import registry
|
|
from lib_py.delegation.service import DelegationService
|
|
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
svc = DelegationService(registry_dir=str(reg_dir))
|
|
jid = svc.submit_job("service prompt", agent="claude")
|
|
rec = svc.load_job(jid)
|
|
valid_token = rec["dispatch_token"]
|
|
|
|
# 1) Stale token passed via expect_dispatch_token -> returns rc=4
|
|
rc_mismatch = svc.publish_event(jid, "completed", detail="done",
|
|
expect_dispatch_token="stale_tok_xyz")
|
|
assert rc_mismatch == 4, f"Expected rc=4 on dispatch token mismatch, got {rc_mismatch}"
|
|
assert svc.load_job(jid)["status"] == "pending"
|
|
|
|
# 2) Stale token passed via data dict -> returns rc=4
|
|
rc_data_mismatch = svc.publish_event(jid, "completed", detail="done",
|
|
data={"dispatch_token": "stale_tok_xyz"})
|
|
assert rc_data_mismatch == 4, f"Expected rc=4 on data dispatch token mismatch, got {rc_data_mismatch}"
|
|
assert svc.load_job(jid)["status"] == "pending"
|
|
|
|
# 3) Valid token -> succeeds and updates status
|
|
rc_valid = svc.publish_event(jid, "completed", detail="done",
|
|
expect_dispatch_token=valid_token)
|
|
assert rc_valid in (0, 2), f"Expected 0/2 for valid token, got {rc_valid}"
|
|
assert svc.load_job(jid)["status"] == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-24: Redispatched job terminal calls require pinned dispatch token
|
|
# (Codex Round 7 finding: all 4 stale/untokened CLI & service paths rejected rc=4)
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_24_redispatched_job_terminal_requires_pinned_dispatch_token(tmp_path):
|
|
from lib_py.delegation import registry, engine
|
|
from lib_py.delegation.job_subscriber import _check_disk_fallback
|
|
from lib_py.delegation.service import DelegationService
|
|
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
pub_py = str(Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py")
|
|
|
|
def _setup_redispatched_job(prefix):
|
|
jid = registry.register_job(f"j_{prefix}", "prompt", "claude", "herdr:claude", registry_dir=str(reg_dir))
|
|
rec1 = registry.load_job(jid, str(reg_dir))
|
|
tok1 = rec1["dispatch_token"]
|
|
# Timeout and redispatch to attempt 2
|
|
engine.transition(jid, "deadline_timeout", registry_dir=str(reg_dir), source="operator")
|
|
engine.transition(jid, "redispatch", registry_dir=str(reg_dir), source="operator", fields={"attempt_id": 2})
|
|
rec2 = registry.load_job(jid, str(reg_dir))
|
|
assert rec2["status"] == "pending"
|
|
tok2 = rec2["dispatch_token"]
|
|
assert tok2 != tok1
|
|
assert rec2["attempt_id"] == 2
|
|
return jid, tok1, tok2
|
|
|
|
# Case 1: CLI with stale token -> rc=4, status remains pending, disk fallback has no terminal
|
|
j1, tok1_1, tok2_1 = _setup_redispatched_job("cli_stale")
|
|
cmd_cli_stale = [
|
|
sys.executable, pub_py, "--registry-dir", str(reg_dir),
|
|
"--job", j1, "--event", "completed", "--detail", "stale cli",
|
|
"--expect-dispatch-token", tok1_1, "--attempt-id", "1"
|
|
]
|
|
res1 = subprocess.run(cmd_cli_stale, capture_output=True, text=True)
|
|
assert res1.returncode == 4
|
|
assert registry.load_job(j1, str(reg_dir))["status"] == "pending"
|
|
p1, t1 = {j1}, {}
|
|
_check_disk_fallback(p1, str(reg_dir), t1, {j1: tok2_1}, {j1: 2})
|
|
assert j1 in p1 and j1 not in t1
|
|
|
|
# Case 2: CLI WITHOUT token/attempt -> rc=4, status remains pending, disk fallback has no terminal
|
|
j2, tok1_2, tok2_2 = _setup_redispatched_job("cli_untokened")
|
|
cmd_cli_untokened = [
|
|
sys.executable, pub_py, "--registry-dir", str(reg_dir),
|
|
"--job", j2, "--event", "completed", "--detail", "untokened cli"
|
|
]
|
|
res2 = subprocess.run(cmd_cli_untokened, capture_output=True, text=True)
|
|
assert res2.returncode == 4, f"Expected rc=4 for untokened CLI on redispatched job, got {res2.returncode}"
|
|
assert registry.load_job(j2, str(reg_dir))["status"] == "pending"
|
|
p2, t2 = {j2}, {}
|
|
_check_disk_fallback(p2, str(reg_dir), t2, {j2: tok2_2}, {j2: 2})
|
|
assert j2 in p2 and j2 not in t2
|
|
|
|
# Case 3: Service with stale token in data -> rc=4, status remains pending, disk fallback has no terminal
|
|
j3, tok1_3, tok2_3 = _setup_redispatched_job("svc_stale")
|
|
svc = DelegationService(registry_dir=str(reg_dir))
|
|
rc3 = svc.publish_event(j3, "completed", detail="stale svc", data={"dispatch_token": tok1_3, "attempt_id": 1})
|
|
assert rc3 == 4
|
|
assert registry.load_job(j3, str(reg_dir))["status"] == "pending"
|
|
p3, t3 = {j3}, {}
|
|
_check_disk_fallback(p3, str(reg_dir), t3, {j3: tok2_3}, {j3: 2})
|
|
assert j3 in p3 and j3 not in t3
|
|
|
|
# Case 4: Service WITHOUT token/attempt -> rc=4, status remains pending, disk fallback has no terminal
|
|
j4, tok1_4, tok2_4 = _setup_redispatched_job("svc_untokened")
|
|
rc4 = svc.publish_event(j4, "completed", detail="untokened svc")
|
|
assert rc4 == 4, f"Expected rc=4 for untokened service on redispatched job, got {rc4}"
|
|
assert registry.load_job(j4, str(reg_dir))["status"] == "pending"
|
|
p4, t4 = {j4}, {}
|
|
_check_disk_fallback(p4, str(reg_dir), t4, {j4: tok2_4}, {j4: 2})
|
|
assert j4 in p4 and j4 not in t4
|
|
|
|
# Legitimate Attempt 2 completion with valid tok2 succeeds
|
|
rc_valid = svc.publish_event(j4, "completed", detail="valid attempt 2", expect_dispatch_token=tok2_4, attempt_id=2)
|
|
assert rc_valid in (0, 2)
|
|
assert registry.load_job(j4, str(reg_dir))["status"] == "completed"
|
|
_check_disk_fallback(p4, str(reg_dir), t4, {j4: tok2_4}, {j4: 2})
|
|
assert j4 not in p4 and t4.get(j4) == "completed"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# T1-25: stop_session pins token/attempt and successfully cancels redispatched job
|
|
# (Codex Round 8 P2: lifecycle stop helper passes pinned token/attempt)
|
|
# ---------------------------------------------------------------------------
|
|
def test_t1_25_stop_session_cancels_redispatched_job_with_pinned_token(tmp_path, monkeypatch):
|
|
from lib_py.delegation import registry, engine
|
|
|
|
reg_dir = tmp_path / "jobs"
|
|
reg_dir.mkdir(parents=True, exist_ok=True)
|
|
monkeypatch.setenv("MAM_JOBS_DIR", str(reg_dir))
|
|
|
|
jid = registry.register_job("job_stop_cas", "prompt stop", "claude", "herdr:claude", registry_dir=str(reg_dir))
|
|
rec1 = registry.load_job(jid, str(reg_dir))
|
|
tok1 = rec1["dispatch_token"]
|
|
assert rec1["attempt_id"] == 1
|
|
|
|
# Timeout & redispatch to attempt 2
|
|
engine.transition(jid, "deadline_timeout", registry_dir=str(reg_dir), source="operator")
|
|
engine.transition(jid, "redispatch", registry_dir=str(reg_dir), source="operator", fields={"attempt_id": 2})
|
|
rec2 = registry.load_job(jid, str(reg_dir))
|
|
tok2 = rec2["dispatch_token"]
|
|
assert rec2["status"] == "pending"
|
|
assert rec2["attempt_id"] == 2
|
|
assert tok2 != tok1
|
|
|
|
# Simulate stop_session pinning token & attempt at stop time
|
|
# 1) Correct pinned token/attempt cancels the redispatched attempt 2 job
|
|
res_cancel = subprocess.run([
|
|
"bash", "-c",
|
|
f'source .agents/skills/lib.sh; delegate_publish_event "{jid}" cancelled "session stopped by operator" "{tok2}" "2"'
|
|
], cwd=str(Path(__file__).parent.parent), capture_output=True, text=True)
|
|
assert res_cancel.returncode == 0
|
|
rec_after = registry.load_job(jid, str(reg_dir))
|
|
assert rec_after["status"] == "cancelled"
|
|
|
|
# 2) Race condition check: if redispatched to attempt 3 before cancellation landed,
|
|
# a stale cancellation with attempt 2 token must NOT cancel attempt 3
|
|
engine.transition(jid, "redispatch", registry_dir=str(reg_dir), source="operator", fields={"attempt_id": 3})
|
|
rec3 = registry.load_job(jid, str(reg_dir))
|
|
tok3 = rec3["dispatch_token"]
|
|
assert rec3["status"] == "pending"
|
|
assert rec3["attempt_id"] == 3
|
|
assert tok3 != tok2
|
|
|
|
# Attempt to cancel with stale tok2 from attempt 2
|
|
pub_py = str(Path(__file__).parent.parent / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "scripts" / "publish_event.py")
|
|
res_stale_cancel = subprocess.run([
|
|
sys.executable, pub_py, "--registry-dir", str(reg_dir),
|
|
"--job", jid, "--event", "cancelled", "--detail", "stale stop",
|
|
"--expect-dispatch-token", tok2, "--attempt-id", "2"
|
|
], capture_output=True, text=True)
|
|
assert res_stale_cancel.returncode == 4
|
|
assert registry.load_job(jid, str(reg_dir))["status"] == "pending"
|