GodopuandClaude Opus 5.5
933af344c2
S11 keeps the generated .mam/shim/herdr from changing by accident; #45 changes
it on purpose, so its base moves to ed73f67.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1296 lines
66 KiB
Python
1296 lines
66 KiB
Python
"""Issue #23 + #24 — the create_session start path.
|
||
|
||
Four regressions live here, in the order of the issues:
|
||
|
||
1. a herdr call aimed at another server must not carry the caller pane's own
|
||
workspace/pane/tab ids (#23 comments: reproduced on Linux and macOS);
|
||
2. blocking start-up modals are agent-adapter knowledge and are answered while
|
||
waiting for the TUI (#24);
|
||
3. onboarding delivery is judged by the agent's own `started` event (#23);
|
||
4. a rollback verifies the kill and leaves no `running` row or dangling job (#23).
|
||
|
||
What is mocked and what is real is stated per test. The dialog tests fix OUR
|
||
contract - which pane string gets which keys - not that the keys drive a real
|
||
agy/grok/codex/opencode TUI; no such TUI can be started here.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import textwrap
|
||
|
||
import pytest
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# helpers
|
||
# --------------------------------------------------------------------------
|
||
|
||
def _run_lib(tmp_path, body, env=None):
|
||
"""Source the sandbox lib.sh and run `body`. Unlike the shim-contract
|
||
helper this keeps HERDR_* exactly as `env` sets it: these tests are about
|
||
what the caller's pane leaks in."""
|
||
lib_path = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
script = 'source "%s"\n_init_herdr_isolation\n%s\n' % (lib_path, body)
|
||
run_env = dict(os.environ)
|
||
for var in ("HERDR_SESSION", "HERDR_SESSION_NAME", "HERDR_WORKSPACE_ID",
|
||
"HERDR_PANE_ID", "HERDR_TAB_ID", "HERDR_SOCKET_PATH"):
|
||
run_env.pop(var, None)
|
||
run_env.update(env or {})
|
||
return subprocess.run(["bash", "-c", script], capture_output=True,
|
||
text=True, cwd=str(tmp_path), env=run_env)
|
||
|
||
|
||
def _mock_state(mock_herdr):
|
||
with open(mock_herdr) as f:
|
||
return json.load(f)
|
||
|
||
|
||
def _write_mock_state(mock_herdr, state):
|
||
with open(mock_herdr, "w") as f:
|
||
json.dump(state, f, indent=2)
|
||
|
||
|
||
def _seed_agent(mock_herdr, tmp_path, name, workspace_id, pane_id):
|
||
state = _mock_state(mock_herdr)
|
||
state.setdefault("workspaces", [])
|
||
if not any(w.get("workspace_id") == workspace_id for w in state["workspaces"]):
|
||
state["workspaces"].append(
|
||
{"workspace_id": workspace_id, "label": workspace_id, "cwd": str(tmp_path)})
|
||
state.setdefault("agents", {})[name] = {
|
||
"name": name, "agent": "agy", "status": "running", "cwd": str(tmp_path),
|
||
"workspace_id": workspace_id, "pane_id": pane_id, "pid": 9999,
|
||
"command": "agy", "buffer": "Ready",
|
||
}
|
||
state.setdefault("panes", []).append(
|
||
{"pane_id": pane_id, "workspace_id": workspace_id, "label": name, "buffer": "Ready"})
|
||
_write_mock_state(mock_herdr, state)
|
||
|
||
|
||
def _install_env_recorder(tmp_path, monkeypatch):
|
||
"""A herdr earlier in PATH than the mock that dumps the env it was called
|
||
with, then execs the mock. The shim resolves the first non-shim herdr on
|
||
PATH, so this observes exactly what reached the real binary."""
|
||
rec_dir = tmp_path / "recorder"
|
||
rec_dir.mkdir(parents=True, exist_ok=True)
|
||
env_log = tmp_path / "herdr_env.ndjson"
|
||
real = tmp_path / "bin" / "herdr"
|
||
script = rec_dir / "herdr"
|
||
script.write_text(textwrap.dedent('''\
|
||
#!/usr/bin/env python3
|
||
import json, os, sys
|
||
with open(%r, "a") as f:
|
||
f.write(json.dumps({
|
||
"argv": sys.argv[1:],
|
||
"HERDR_WORKSPACE_ID": os.environ.get("HERDR_WORKSPACE_ID"),
|
||
"HERDR_PANE_ID": os.environ.get("HERDR_PANE_ID"),
|
||
"HERDR_TAB_ID": os.environ.get("HERDR_TAB_ID"),
|
||
"HERDR_SESSION": os.environ.get("HERDR_SESSION"),
|
||
"HERDR_SESSION_NAME": os.environ.get("HERDR_SESSION_NAME"),
|
||
}) + "\\n")
|
||
os.execv(%r, [%r] + sys.argv[1:])
|
||
''') % (str(env_log), str(real), str(real)))
|
||
script.chmod(0o755)
|
||
monkeypatch.setenv("PATH", "%s:%s" % (rec_dir, os.environ["PATH"]))
|
||
return env_log
|
||
|
||
|
||
def _recorded(env_log):
|
||
if not os.path.exists(env_log):
|
||
return []
|
||
return [json.loads(l) for l in open(env_log) if l.strip()]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# item 1 — caller pane scope must not leak across servers
|
||
# --------------------------------------------------------------------------
|
||
|
||
CROSS = {"HERDR_SESSION": "default", "HERDR_SESSION_NAME": "mam-isolated",
|
||
"HERDR_WORKSPACE_ID": "w5", "HERDR_PANE_ID": "w5:p2",
|
||
"HERDR_TAB_ID": "w5:t1"}
|
||
|
||
# The orchestrator pane as the live trial of this branch measured it (herdr
|
||
# 0.8.2): no HERDR_SESSION at all, the default server's socket, and that
|
||
# server's own ids. This is the env in which the HERDR_SESSION-based verdict
|
||
# was always "not crossing" and #23 reproduced unchanged.
|
||
DEFAULT_SOCK = "/home/someone/.config/herdr/herdr.sock"
|
||
NAMED_SOCK = "/home/someone/.config/herdr/sessions/%s/herdr.sock"
|
||
TRIAL_PANE = {"HERDR_SOCKET_PATH": DEFAULT_SOCK, "HERDR_WORKSPACE_ID": "w6",
|
||
"HERDR_PANE_ID": "w6:p1", "HERDR_TAB_ID": "w6:t1",
|
||
"HERDR_SESSION_NAME": "multi-agent-mux"}
|
||
|
||
|
||
def test_s1_cross_server_call_drops_the_caller_panes_scope_ids(
|
||
mam_sandbox, mock_herdr, mock_agents, monkeypatch):
|
||
"""S1 (mock herdr): the caller sits in another server's pane. has-session,
|
||
kill-session and agent prompt must find the agent, and the three ids must
|
||
not reach herdr."""
|
||
tmp_path = mam_sandbox
|
||
env_log = _install_env_recorder(tmp_path, monkeypatch)
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
|
||
res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01', env=CROSS)
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
|
||
res = _run_lib(tmp_path, 'herdr agent prompt creator-agy-01 "hello"', env=CROSS)
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
|
||
res = _run_lib(tmp_path, 'herdr kill-session -t creator-agy-01', env=CROSS)
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
|
||
calls = _recorded(env_log)
|
||
assert calls, "the recorder saw no herdr call"
|
||
for call in calls:
|
||
assert call["HERDR_WORKSPACE_ID"] is None, call
|
||
assert call["HERDR_PANE_ID"] is None, call
|
||
assert call["HERDR_TAB_ID"] is None, call
|
||
|
||
# the kill really happened, i.e. this is not a silent no-op
|
||
assert "creator-agy-01" not in _mock_state(mock_herdr).get("agents", {})
|
||
|
||
|
||
def test_s1b_the_delegate_job_wrappers_bare_herdr_is_the_lib_sh_function(
|
||
mam_sandbox, mock_herdr, mock_agents, monkeypatch):
|
||
"""S1b (mock herdr): the wrapper sources lib.sh and then calls a bare
|
||
`herdr` (wrapper :918 preflight). A shell function beats PATH, so that call
|
||
is mam_herdr and is fixed by item 1 too - this is why #23's preflight
|
||
failure is cured by the same change."""
|
||
tmp_path = mam_sandbox
|
||
_install_env_recorder(tmp_path, monkeypatch)
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
|
||
res = _run_lib(tmp_path, 'echo "kind=$(type -t herdr)"\nherdr has-session -t creator-agy-01',
|
||
env=CROSS)
|
||
assert "kind=function" in res.stdout, res.stdout + res.stderr
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
|
||
|
||
def test_s2_same_server_keeps_the_explicit_workspace_scope(
|
||
mam_sandbox, mock_herdr, mock_agents, monkeypatch):
|
||
"""S2 (mock herdr): on the target's own server the exported scope still
|
||
scopes - a same-named agent in another workspace is not addressed (H-22)."""
|
||
tmp_path = mam_sandbox
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
same = {"HERDR_SESSION": "mam-isolated", "HERDR_SESSION_NAME": "mam-isolated",
|
||
"HERDR_WORKSPACE_ID": "w2"}
|
||
|
||
before = len(_mock_state(mock_herdr).get("calls", []))
|
||
res = _run_lib(tmp_path, 'herdr agent prompt creator-agy-01 "hello from w2"', env=same)
|
||
assert res.returncode != 0, res.stdout + res.stderr
|
||
new = _mock_state(mock_herdr).get("calls", [])[before:]
|
||
prompts = [c for c in new if len(c) >= 2 and c[0] == "agent" and c[1] == "prompt"]
|
||
assert prompts == [], new
|
||
|
||
|
||
def test_s2b_a_caller_outside_any_pane_keeps_the_explicit_scope(
|
||
mam_sandbox, mock_herdr, mock_agents):
|
||
"""S2b (mock herdr): no HERDR_SESSION means the caller is not in a pane at
|
||
all, so an exported HERDR_WORKSPACE_ID is a deliberate scope, not a leak.
|
||
This is the case H-18/H-21/H-23 fix."""
|
||
tmp_path = mam_sandbox
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
|
||
res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01',
|
||
env={"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"})
|
||
assert res.returncode == 1, "an explicit out-of-scope id must still hide the agent"
|
||
|
||
res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01',
|
||
env={"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w1"})
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
|
||
|
||
def test_s1c_the_real_default_server_pane_is_a_crossing(
|
||
mam_sandbox, mock_herdr, mock_agents, monkeypatch):
|
||
"""S1c (trial finding, req. 4): the caller's server comes from the socket
|
||
path when HERDR_SESSION is absent. In the measured orchestrator pane the
|
||
three ids must be dropped and the agent on the named server found."""
|
||
tmp_path = mam_sandbox
|
||
env_log = _install_env_recorder(tmp_path, monkeypatch)
|
||
_seed_agent(mock_herdr, tmp_path, "trial-agy", "w1", "w1:p5")
|
||
|
||
res = _run_lib(tmp_path, 'herdr has-session -t trial-agy', env=TRIAL_PANE)
|
||
assert res.returncode == 0, "the trial's exact failure: " + res.stdout + res.stderr
|
||
|
||
calls = _recorded(env_log)
|
||
assert calls, "the recorder saw no herdr call"
|
||
for call in calls:
|
||
assert call["HERDR_WORKSPACE_ID"] is None, call
|
||
assert call["HERDR_PANE_ID"] is None, call
|
||
assert call["HERDR_TAB_ID"] is None, call
|
||
|
||
|
||
def test_s1d_the_verdict_itself_on_the_trial_pane(mam_sandbox):
|
||
"""S1d: the function the trial printed, in the trial's env: crossing."""
|
||
res = _run_lib(mam_sandbox,
|
||
'echo "caller=$(_herdr_caller_server)"\n'
|
||
'if _herdr_call_is_cross_server; then echo VERDICT=crossing; '
|
||
'else echo VERDICT=not-crossing; fi',
|
||
env=TRIAL_PANE)
|
||
assert "caller=default" in res.stdout, res.stdout + res.stderr
|
||
assert "VERDICT=crossing" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
def test_s1e_a_named_server_pane_calling_its_own_server_keeps_its_scope(
|
||
mam_sandbox, mock_herdr, mock_agents):
|
||
"""S1e (req. 4): a caller in a named server's pane (socket under
|
||
sessions/<name>/) targeting that same server is not a crossing - the
|
||
exported workspace id still scopes, as in S2."""
|
||
tmp_path = mam_sandbox
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
same = {"HERDR_SOCKET_PATH": NAMED_SOCK % "mam-isolated",
|
||
"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"}
|
||
res = _run_lib(tmp_path, 'echo "caller=$(_herdr_caller_server)"\nherdr has-session -t creator-agy-01',
|
||
env=same)
|
||
assert "caller=mam-isolated" in res.stdout, res.stdout + res.stderr
|
||
assert res.returncode == 1, "same server: the explicit w2 scope must still hide a w1 agent"
|
||
|
||
|
||
def test_s1f_a_named_server_pane_calling_another_named_server_is_a_crossing(
|
||
mam_sandbox, mock_herdr, mock_agents):
|
||
"""S1f: named socket, different named target, no HERDR_SESSION - crossing."""
|
||
tmp_path = mam_sandbox
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
other = {"HERDR_SOCKET_PATH": NAMED_SOCK % "mam-other",
|
||
"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"}
|
||
res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01', env=other)
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
|
||
|
||
def test_s1g_herdr_session_wins_over_the_socket_when_both_are_present(mam_sandbox):
|
||
"""S1g (req. 1): HERDR_SESSION, when a herdr exports it, is authoritative
|
||
over the socket-derived name."""
|
||
res = _run_lib(mam_sandbox, 'echo "caller=$(_herdr_caller_server)"',
|
||
env={"HERDR_SESSION": "from-herdr", "HERDR_SOCKET_PATH": DEFAULT_SOCK})
|
||
assert "caller=from-herdr" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
def test_s1h_an_unrecognised_socket_layout_is_not_in_a_pane(mam_sandbox, mock_herdr, mock_agents):
|
||
"""S1h: a socket path that is neither layout yields no caller server, so
|
||
an exported scope is respected (today's behaviour), never guessed away."""
|
||
tmp_path = mam_sandbox
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
res = _run_lib(tmp_path, 'echo "caller=[$(_herdr_caller_server)]"\nherdr has-session -t creator-agy-01',
|
||
env={"HERDR_SOCKET_PATH": "/tmp/odd/herdr.sock",
|
||
"HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"})
|
||
assert "caller=[]" in res.stdout, res.stdout + res.stderr
|
||
assert res.returncode == 1, res.stdout + res.stderr
|
||
|
||
|
||
def test_s3_shim_contract_regressions_still_pass(mam_sandbox):
|
||
"""S3: H-18 / H-21 / H-23 are the tests that fix today's scoping meaning.
|
||
Run them explicitly as this change's regression guard."""
|
||
repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
res = subprocess.run(
|
||
[sys.executable, "-m", "pytest", "-q",
|
||
"-p", "no:cacheprovider",
|
||
"tests/test_herdr_shim_contract.py", "-k",
|
||
"h18 or h21 or h22 or h23"],
|
||
capture_output=True, text=True, cwd=repo,
|
||
env=dict(os.environ, PYTHONDONTWRITEBYTECODE="1"))
|
||
assert res.returncode == 0, res.stdout[-4000:] + res.stderr[-2000:]
|
||
|
||
|
||
def test_s11_this_branch_does_not_change_the_generated_shim(mam_sandbox):
|
||
"""S11: the item-1 fix lives in mam_herdr, never in the heredoc that
|
||
lib.sh materialises into $WORKSPACE_ROOT/.mam/shim/herdr. That file is
|
||
shared by every checkout bound to one control plane, so keeping it
|
||
byte-identical to base is what makes "the control plane is not modified"
|
||
true at runtime. Issue #45 changed the heredoc on purpose (the bootstrap
|
||
starts the server with restore-on-restart off), so the base is that commit."""
|
||
repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
base = os.environ.get("MAM_BASE_SHA", "ed73f67")
|
||
show = subprocess.run(["git", "show", "%s:.agents/skills/lib.sh" % base],
|
||
capture_output=True, text=True, cwd=repo)
|
||
if show.returncode != 0:
|
||
pytest.skip("base %s not available in this checkout" % base)
|
||
with open(os.path.join(repo, ".agents", "skills", "lib.sh")) as f:
|
||
head = f.read()
|
||
assert _heredoc_body(head) == _heredoc_body(show.stdout), (
|
||
"the shim heredoc changed: the generated .mam/shim/herdr of this branch "
|
||
"would differ from the control root's, and the two would overwrite "
|
||
"each other on every herdr call")
|
||
|
||
|
||
def _heredoc_body(text):
|
||
lines = text.splitlines(True)
|
||
start = None
|
||
for i, line in enumerate(lines):
|
||
if line.strip() == "cat <<'EOF' > \"$tmp_file\"":
|
||
start = i + 1
|
||
break
|
||
assert start is not None, "shim heredoc not found in lib.sh"
|
||
for j in range(start, len(lines)):
|
||
if lines[j].rstrip("\n") == "EOF":
|
||
return "".join(lines[start:j])
|
||
raise AssertionError("unterminated shim heredoc")
|
||
|
||
|
||
@pytest.mark.parametrize("env,expected_rc", [(CROSS, 1), ({"HERDR_SESSION_NAME": "mam-isolated"}, 1)])
|
||
def test_s12_mam_herdr_never_aborts_its_caller_under_set_e(
|
||
mam_sandbox, mock_herdr, mock_agents, env, expected_rc):
|
||
"""S12 (mock herdr): mam_herdr runs inside scripts using `set -euo
|
||
pipefail`. Neither branch of the cross-server test may abort the shell,
|
||
and the child's rc must be the function's rc."""
|
||
tmp_path = mam_sandbox
|
||
body = ('set -euo pipefail\n'
|
||
'rc=0\n'
|
||
'mam_herdr has-session -t nobody-here || rc=$?\n'
|
||
'echo "missing_rc=$rc"\n'
|
||
'_seen=0\n'
|
||
'mam_herdr has-session -t creator-agy-01 || _seen=$?\n'
|
||
'echo "present_rc=$_seen"\n'
|
||
'echo REACHED_END\n')
|
||
_seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
|
||
lib_path = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
script = 'source "%s"\n%s' % (lib_path, body)
|
||
run_env = dict(os.environ)
|
||
for var in ("HERDR_SESSION", "HERDR_SESSION_NAME", "HERDR_WORKSPACE_ID",
|
||
"HERDR_PANE_ID", "HERDR_TAB_ID"):
|
||
run_env.pop(var, None)
|
||
run_env.update(env)
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=run_env)
|
||
assert "REACHED_END" in res.stdout, res.stdout + res.stderr
|
||
assert "missing_rc=%d" % expected_rc in res.stdout, res.stdout + res.stderr
|
||
assert "present_rc=0" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
def test_s13_the_loop_watcher_applies_the_same_rule(mam_sandbox):
|
||
"""S13 (pure unit): watcher.py builds its own herdr argv and never goes
|
||
through mam_herdr, so it needs the same rule."""
|
||
repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, os.path.join(repo, ".agents", "skills"))
|
||
from lib_py.loop import watcher
|
||
|
||
caller_pane = {"HERDR_SESSION": "default", "HERDR_WORKSPACE_ID": "w5",
|
||
"HERDR_PANE_ID": "w5:p2", "HERDR_TAB_ID": "w5:t1", "PATH": "/usr/bin"}
|
||
|
||
cross = watcher.scoped_env("mam-isolated", environ=caller_pane)
|
||
assert "HERDR_WORKSPACE_ID" not in cross and "HERDR_PANE_ID" not in cross
|
||
assert "HERDR_TAB_ID" not in cross and cross["PATH"] == "/usr/bin"
|
||
|
||
same = watcher.scoped_env("default", environ=caller_pane)
|
||
assert same["HERDR_WORKSPACE_ID"] == "w5"
|
||
|
||
outside = watcher.scoped_env("mam-isolated", environ={"HERDR_WORKSPACE_ID": "w5"})
|
||
assert outside["HERDR_WORKSPACE_ID"] == "w5"
|
||
|
||
none_target = watcher.scoped_env(None, environ=caller_pane)
|
||
assert none_target["HERDR_WORKSPACE_ID"] == "w5"
|
||
|
||
# trial finding: no HERDR_SESSION, the caller's server comes from the socket
|
||
trial = {"HERDR_SOCKET_PATH": DEFAULT_SOCK, "HERDR_WORKSPACE_ID": "w6",
|
||
"HERDR_PANE_ID": "w6:p1", "HERDR_TAB_ID": "w6:t1"}
|
||
assert watcher.caller_server(trial) == "default"
|
||
assert "HERDR_WORKSPACE_ID" not in watcher.scoped_env("multi-agent-mux", environ=trial)
|
||
named_same = {"HERDR_SOCKET_PATH": NAMED_SOCK % "iso-x", "HERDR_WORKSPACE_ID": "w2"}
|
||
assert watcher.caller_server(named_same) == "iso-x"
|
||
assert watcher.scoped_env("iso-x", environ=named_same)["HERDR_WORKSPACE_ID"] == "w2"
|
||
assert "HERDR_WORKSPACE_ID" not in watcher.scoped_env("iso-y", environ=named_same)
|
||
assert watcher.caller_server({"HERDR_SOCKET_PATH": "/tmp/odd/herdr.sock"}) == ""
|
||
assert watcher.caller_server({"HERDR_SESSION": "s", "HERDR_SOCKET_PATH": DEFAULT_SOCK}) == "s"
|
||
|
||
seen = {}
|
||
|
||
def _run(argv, **kw):
|
||
seen["env"] = kw.get("env")
|
||
|
||
class P:
|
||
returncode, stdout, stderr = 0, "{}", ""
|
||
return P()
|
||
|
||
watcher.wait_once("agt", session="iso-x", run=_run)
|
||
assert seen["env"] is not None and "HERDR_SESSION" not in ("",)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# item 2 — start-up dialogs come from the adapter (#24)
|
||
#
|
||
# MOCKED, and this is the limit of it: the pane text is a fixture string and
|
||
# send-keys is a shell stub, so these fix OUR contract - which screen signature
|
||
# gets which keys, once. They do NOT show that the keys drive a real agy, grok,
|
||
# codex or opencode TUI; no such TUI can be started in this environment. That
|
||
# check is a live herdr run after the loop.
|
||
# --------------------------------------------------------------------------
|
||
|
||
def _run_dialogs(tmp_path, agent, pane_text, timeout=1, env=None, prelude=""):
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
pane_file = tmp_path / "pane.txt"
|
||
pane_file.write_text(pane_text)
|
||
keys_log = tmp_path / "sent_keys.txt"
|
||
if keys_log.exists():
|
||
keys_log.unlink()
|
||
script = '''
|
||
source "%s"
|
||
_pane_tail() { cat "%s"; }
|
||
_sks_herdr() { printf '%%s\\n' "$*" >> "%s"; }
|
||
%s
|
||
handle_startup_dialogs test-sess %d %s
|
||
echo "rc=$?"
|
||
''' % (lib, pane_file, keys_log, prelude, timeout, agent)
|
||
run_env = dict(os.environ)
|
||
run_env.pop("MAM_AUTO_TRUST", None)
|
||
for k, v in (env or {}).items():
|
||
if v == "":
|
||
run_env.pop(k, None) # "" means: not in the environment at all
|
||
else:
|
||
run_env[k] = v
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=run_env)
|
||
sent = keys_log.read_text().splitlines() if keys_log.exists() else []
|
||
return res, sent
|
||
|
||
|
||
AGENT_DIALOGS = [
|
||
("agy", "Do you trust the contents of this project?\n > Yes, I trust this folder",
|
||
["send-keys -t test-sess Enter"]),
|
||
("grok", "Do you trust the contents of this directory?",
|
||
["send-keys -t test-sess y"]),
|
||
("codex", "Do you trust the contents of this directory?\n > 1. Yes, continue",
|
||
["send-keys -t test-sess Enter"]),
|
||
("codex", "Update available\n 1. Update now\n 2. Skip\n 3. Skip until next version",
|
||
["send-keys -t test-sess 2"]),
|
||
("claude", "Do you trust the files in this folder?\n Yes, I trust this folder",
|
||
["send-keys -t test-sess Enter"]),
|
||
("claude", "Try the new fullscreen renderer?\n Yes, try it",
|
||
["send-keys -t test-sess Escape"]),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("agent,pane,expected", AGENT_DIALOGS)
|
||
def test_s4_each_agents_dialog_signature_gets_its_own_keys(
|
||
mam_sandbox, agent, pane, expected):
|
||
"""S4 (stubbed pane + stubbed send-keys): the adapter's signature is
|
||
matched and exactly its keys are sent."""
|
||
res, sent = _run_dialogs(mam_sandbox, agent, pane)
|
||
assert sent == expected, (res.stdout, res.stderr, sent)
|
||
|
||
|
||
def test_s4b_every_adapter_publishes_the_fact(mam_sandbox):
|
||
"""S4b (real `lib_py.agents facts`): MAM_STARTUP_DIALOGS is part of the
|
||
facts contract for every agent, so the shell never has to know an agent
|
||
name. An agent with no known blocking modal publishes an empty list."""
|
||
repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
skills = os.path.join(repo, ".agents", "skills")
|
||
seen = {}
|
||
for agent in ("claude", "agy", "grok", "codex", "opencode", "hermes"):
|
||
out = subprocess.run(
|
||
[sys.executable, "-m", "lib_py.agents",
|
||
"facts", agent],
|
||
capture_output=True, text=True, cwd=repo,
|
||
env=dict(os.environ, PYTHONPATH=skills)).stdout
|
||
line = [l for l in out.splitlines() if l.startswith("MAM_STARTUP_DIALOGS=")]
|
||
assert line, "no MAM_STARTUP_DIALOGS for %s" % agent
|
||
seen[agent] = json.loads(line[0].split("=", 1)[1].strip("'"))
|
||
for agent in ("claude", "agy", "grok", "codex"):
|
||
assert seen[agent], "%s must publish its start-up dialogs" % agent
|
||
for pattern, keys, kind in seen[agent]:
|
||
assert kind in ("trust", "notice"), (agent, kind)
|
||
assert keys, (agent, pattern)
|
||
assert seen["opencode"] == [] and seen["hermes"] == []
|
||
|
||
|
||
def test_s5_an_accepted_dialog_still_on_screen_is_not_resent(mam_sandbox):
|
||
"""S5 (stubbed pane + stubbed send-keys): #24's comment reports that
|
||
`--source visible` keeps showing the dialog for a while after it was
|
||
accepted, and _pane_capture hardcodes that source. The pane here never
|
||
changes, so a frame-count rule would send the keys again every poll; the
|
||
time cooldown must keep it at one."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "grok", "Do you trust the contents of this directory?",
|
||
timeout=3, env={"MAM_DIALOG_RESEND_S": "60"})
|
||
assert sent == ["send-keys -t test-sess y"], (res.stdout, res.stderr, sent)
|
||
|
||
|
||
def test_s5b_the_cooldown_expiring_allows_one_more_try(mam_sandbox):
|
||
"""S5b: the cooldown is a rate limit, not a one-shot - a dialog that really
|
||
did not take the keys is retried once the window passes."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "grok", "Do you trust the contents of this directory?",
|
||
timeout=4, env={"MAM_DIALOG_RESEND_S": "1"})
|
||
assert len(sent) >= 2, (res.stdout, res.stderr, sent)
|
||
assert set(sent) == {"send-keys -t test-sess y"}, sent
|
||
|
||
|
||
def test_s5c_readiness_is_judged_by_the_ready_tokens_not_by_the_dialog_leaving(
|
||
mam_sandbox):
|
||
"""S5c: a ready screen returns 0 without sending anything."""
|
||
res, sent = _run_dialogs(mam_sandbox, "grok", "Grok ready\n>>> ", timeout=3)
|
||
assert sent == [], sent
|
||
assert "rc=0" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
READY_FRAMES = [
|
||
("claude", "Welcome to Claude Code\n❯ "),
|
||
("agy", "Antigravity Ready in /tmp\n> "),
|
||
("grok", "Grok Ready\n>>> "),
|
||
("codex", "Codex\nAsk Codex to do anything"),
|
||
("opencode", "OpenCode\nAsk anything"),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("agent,frame", READY_FRAMES)
|
||
def test_s5d_a_ready_first_frame_returns_within_one_iteration_with_no_env(
|
||
mam_sandbox, agent, frame):
|
||
"""S5d (review B-1): called on its own - as resume_session.sh calls it,
|
||
with no wait_for_tui_ready before it and no MAM_*READY* in the environment
|
||
- the handler must recognise the adapter's ready banner on the first frame
|
||
and return at once, not idle through its timeout. `sleep` is stubbed so
|
||
the iteration count is observable and the run is fast; base returned in
|
||
0.03s, the version under review took the full timeout."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, agent, frame, timeout=20,
|
||
env={"MAM_STRONG_READY_TOKENS": "", "MAM_READY_TOKENS": "", "MAM_WEAK_READY_TOKENS": ""},
|
||
prelude='_iters=0\nsleep() { _iters=$((_iters + 1)); }\n'
|
||
'trap \'echo "iterations=$_iters"\' EXIT')
|
||
assert "rc=0" in res.stdout, res.stdout + res.stderr
|
||
assert sent == [], sent
|
||
iters = int(res.stdout.split("iterations=")[1].split()[0])
|
||
assert iters <= 1, "the handler idled %d iterations on a ready screen" % iters
|
||
|
||
|
||
def test_s5e_caller_set_ready_tokens_are_not_overwritten(mam_sandbox):
|
||
"""S5e: the gap-fill must not clobber tokens a caller set on purpose -
|
||
wait_for_tui_ready's own evaluation, or a test steering readiness."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "grok", "Grok Ready\n>>> ", timeout=2,
|
||
env={"MAM_STRONG_READY_TOKENS": "NonExistentStrong", "MAM_READY_TOKENS": "NonExistentStrong"},
|
||
prelude='_iters=0\nsleep() { _iters=$((_iters + 1)); }\n'
|
||
'trap \'echo "iterations=$_iters"\' EXIT')
|
||
iters = int(res.stdout.split("iterations=")[1].split()[0])
|
||
assert iters >= 2, "the caller's tokens were replaced by the adapter's: " + res.stdout
|
||
|
||
|
||
def test_s6_no_auto_trust_leaves_trust_dialogs_alone_but_answers_notices(
|
||
mam_sandbox):
|
||
"""S6 (stubbed): --no-auto-trust. codex trust is left to a human with a
|
||
warning; the update notice still gets answered, since skipping it changes
|
||
nothing and it blocks the session just as hard."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "codex", "Do you trust the contents of this directory?\n > 1. Yes, continue",
|
||
timeout=1, env={"MAM_AUTO_TRUST": "0"})
|
||
assert sent == [], sent
|
||
assert "no-auto-trust" in res.stderr, res.stderr
|
||
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "codex", "Update available\n 1. Update now\n 3. Skip until next version",
|
||
timeout=1, env={"MAM_AUTO_TRUST": "0"})
|
||
assert sent == ["send-keys -t test-sess 2"], (res.stderr, sent)
|
||
|
||
|
||
def test_s6c_a_skipped_trust_dialog_does_not_hide_the_notice_below_it(mam_sandbox):
|
||
"""S6c (merge review, codex probe): with --no-auto-trust and a codex screen
|
||
that shows the trust prompt and the update notice together, the trust
|
||
prompt is skipped and recorded, and the skip's cooldown must not stop the
|
||
picker from answering the notice on the next poll."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "codex",
|
||
"Do you trust the contents of this directory?\n > 1. Yes, continue\n"
|
||
"Update available\n 1. Update now\n 2. Skip\n 3. Skip until next version",
|
||
timeout=3, env={"MAM_AUTO_TRUST": "0", "MAM_DIALOG_RESEND_S": "30"})
|
||
assert sent == ["send-keys -t test-sess 2"], (res.stderr, sent)
|
||
assert "no-auto-trust" in res.stderr, res.stderr
|
||
|
||
|
||
def test_s5f_a_dialog_inside_its_cooldown_does_not_block_the_next_one(mam_sandbox):
|
||
"""S5f (merge review, codex probe): two dialogs on one screen, auto-trust on.
|
||
The trust prompt is answered first; while it is rate-limited the notice
|
||
must still be answered, once each."""
|
||
res, sent = _run_dialogs(
|
||
mam_sandbox, "codex",
|
||
"Do you trust the contents of this directory?\n > 1. Yes, continue\n"
|
||
"Update available\n 1. Update now\n 2. Skip\n 3. Skip until next version",
|
||
timeout=3, env={"MAM_DIALOG_RESEND_S": "60"})
|
||
assert sent == ["send-keys -t test-sess Enter", "send-keys -t test-sess 2"], (res.stderr, sent)
|
||
|
||
|
||
def test_s6b_create_session_no_longer_gates_dialogs_on_claude(mam_sandbox):
|
||
"""S6b (source read): the #24 root cause was the `AGENT = claude` gate at
|
||
the create_session call site, so the gate's absence is the regression."""
|
||
repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
src = open(os.path.join(repo, ".agents", "skills", "multi-agent-mux-create",
|
||
"scripts", "create_session.sh")).read()
|
||
call = [l for l in src.splitlines() if "handle_startup_dialogs" in l]
|
||
assert call, "create_session no longer answers start-up dialogs at all"
|
||
assert all('"$AGENT"' in l for l in call), call
|
||
assert 'if [ "$AGENT" = "claude" ] && [ -z "$LAST_VISIBLE_STATUS_OVERRIDE" ]' not in src
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# item 3 — onboarding delivery is judged by the agent's own `started` (#23)
|
||
# --------------------------------------------------------------------------
|
||
|
||
def _create_sh(tmp_path):
|
||
return tmp_path / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
|
||
|
||
|
||
def _extract(path, start_marker, end_marker, close=None):
|
||
"""Lift a block out of create_session.sh so the test runs the shipped code
|
||
rather than a paraphrase of it. A moved marker fails loudly here.
|
||
|
||
`close` closes the construct the end marker opens: the block runs to the
|
||
first line equal to it after the end marker (e.g. the `fi` of an `if`)."""
|
||
lines = open(path).read().splitlines(True)
|
||
start = next((i for i, l in enumerate(lines) if start_marker in l), None)
|
||
assert start is not None, "start marker %r is gone from %s" % (start_marker, path)
|
||
end = next((i for i, l in enumerate(lines[start:], start) if end_marker in l), None)
|
||
assert end is not None, "end marker %r is gone from %s" % (end_marker, path)
|
||
if close is not None:
|
||
end = next((i for i, l in enumerate(lines[end:], end) if l.strip() == close), None)
|
||
assert end is not None, "block opened at %r is never closed with %r" % (end_marker, close)
|
||
return "".join(lines[start:end + 1])
|
||
|
||
|
||
SKILLS = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
".agents", "skills")
|
||
REGISTRY_PY = os.path.join(SKILLS, "multi-agent-mux-delegate-job", "scripts", "registry.py")
|
||
PUBLISH_PY = os.path.join(SKILLS, "multi-agent-mux-delegate-job", "scripts", "publish_event.py")
|
||
|
||
|
||
def _producer_env(tmp_path):
|
||
env = dict(os.environ)
|
||
env.pop("MAM_ENV_FILE", None) # PoC mode, no broker: local-first is enough
|
||
env["HOME"] = str(tmp_path)
|
||
env["DELEGATE_JOB_LOGS_DIR"] = str(tmp_path / ".mam" / "delegate_job_logs")
|
||
return env
|
||
|
||
|
||
def _seed_job(tmp_path, job_id="job1", status="pending", started_token=None,
|
||
publish_args=None):
|
||
"""A job record made by the real registry and, when asked, a `started`
|
||
published by the REAL publish_event.py - never a hand-written line.
|
||
|
||
Trial 2 found the previous fixture writing `dispatch_token` into the
|
||
started line itself, a field the real producer only writes when its
|
||
command line carries --expect-dispatch-token. A fake producer that differs
|
||
from the real one let five reviewers pass a waiter no real agent could
|
||
satisfy. `started_token` here is what the *command line* passes; the line
|
||
is whatever publish_event makes of it. `publish_args` runs an arbitrary
|
||
publish instead (e.g. the old create_session command shape).
|
||
"""
|
||
jobs = tmp_path / ".mam" / "jobs"
|
||
jobs.mkdir(parents=True, exist_ok=True)
|
||
# a seed is a fresh record: the real registry refuses to re-register an
|
||
# existing id, and several cases seed the same id more than once
|
||
for stale in (jobs / ("%s.json" % job_id), jobs / ("%s.events.log" % job_id)):
|
||
if stale.exists():
|
||
stale.unlink()
|
||
env = _producer_env(tmp_path)
|
||
reg = subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
|
||
"--prompt", "onboarding", "--agent", "claude-code",
|
||
"--agent-session", "herdr:x", "--dispatch-token", "tok-1",
|
||
"--job-id", job_id, "--ack-timeout", "2"],
|
||
capture_output=True, text=True, env=env)
|
||
assert reg.returncode == 0, reg.stdout + reg.stderr
|
||
if started_token is not None or publish_args is not None:
|
||
cmd = [sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", job_id,
|
||
"--event", "started"]
|
||
if publish_args is not None:
|
||
cmd += list(publish_args)
|
||
else:
|
||
cmd += ["--expect-dispatch-token", started_token, "--attempt-id", "1"]
|
||
pub = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=60)
|
||
# a mismatched token is refused by the producer's own CAS check; that
|
||
# is fine here - the point is what (if anything) reached the log
|
||
if started_token in (None, "tok-1"):
|
||
assert pub.returncode == 0, pub.stdout + pub.stderr
|
||
if status != "pending":
|
||
rec = jobs / ("%s.json" % job_id)
|
||
d = json.loads(rec.read_text())
|
||
d["status"] = status
|
||
rec.write_text(json.dumps(d))
|
||
return jobs
|
||
|
||
|
||
def _run_delivery(tmp_path, inject_rc, job_id="job1", seed_events=True, **seed):
|
||
"""Run create_session.sh's own delivery block with a stubbed injection."""
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
block = _extract(_create_sh(tmp_path), "# Inject instructions into the herdr pane",
|
||
"delegate_wait_started", close="fi")
|
||
if seed_events:
|
||
_seed_job(tmp_path, job_id, **seed)
|
||
pub_log = tmp_path / "published.txt"
|
||
script = '''
|
||
set -euo pipefail
|
||
source "%s"
|
||
SESSION_NAME=test-sess
|
||
DELEGATE_JOB_ID=%s
|
||
DELEGATE_JOB_TOKEN=tok-1
|
||
DELEGATE_JOB_ATTEMPT=1
|
||
DELEGATE_JOB_ACK_S=2
|
||
instructions="hello"
|
||
inject_instructions() { return %d; }
|
||
delegate_publish_event() { printf '%%s\\n' "$2" >> "%s"; }
|
||
%s
|
||
echo DELIVERED
|
||
''' % (lib, job_id, inject_rc, pub_log, block)
|
||
env = dict(os.environ, MAM_JOBS_DIR=str(tmp_path / ".mam" / "jobs"))
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=env)
|
||
published = pub_log.read_text().split() if pub_log.exists() else []
|
||
return res, published
|
||
|
||
|
||
def test_s7_create_session_no_longer_publishes_started_for_the_agent(mam_sandbox):
|
||
"""S7 (source): create_session used to publish `started` itself, so the
|
||
registry showed a started job even when the agent never saw the prompt.
|
||
With that line present no started-based verdict can mean anything."""
|
||
src = open(_create_sh(mam_sandbox)).read()
|
||
assert 'delegate_publish_event "$DELEGATE_JOB_ID" started' not in src
|
||
assert "delegate_wait_started" in src
|
||
|
||
|
||
def test_s7b_delivery_succeeds_on_the_agents_own_started(mam_sandbox):
|
||
"""S7b (real registry files, stubbed injection): a started line carrying
|
||
this dispatch token is the evidence."""
|
||
res, published = _run_delivery(mam_sandbox, 0, started_token="tok-1")
|
||
assert "DELIVERED" in res.stdout, res.stdout + res.stderr
|
||
assert published == [], published
|
||
|
||
|
||
def test_s7c_a_started_that_was_followed_by_a_terminal_status_still_counts(mam_sandbox):
|
||
"""S7c: a job that started and then finished (or failed) fast still shows
|
||
its started line - the evidence is the line, whatever the status became."""
|
||
for status in ("running", "completed", "error"):
|
||
res, _ = _run_delivery(mam_sandbox, 0, status=status, started_token="tok-1")
|
||
assert "DELIVERED" in res.stdout, (status, res.stdout + res.stderr)
|
||
|
||
|
||
@pytest.mark.parametrize("status", ["error", "cancelled"])
|
||
def test_s7f_a_job_closed_by_someone_else_is_not_an_acknowledgement(mam_sandbox, status):
|
||
"""S7f (review R1 + req. 4): a record can leave `pending` because another
|
||
writer closed it. That is not evidence - the wait runs to its deadline
|
||
rather than returning early (the 'no started event naming this dispatch'
|
||
path, never the DELIVERED-without-warning path of S7c) - and, the job
|
||
being already terminal, nothing is published over it (req. 4)."""
|
||
res, published = _run_delivery(mam_sandbox, 0, status=status)
|
||
assert published == [], (status, published, res.stderr)
|
||
assert "no started event naming this dispatch" in res.stderr, res.stderr
|
||
assert "already '%s'" % status in res.stderr, res.stderr
|
||
|
||
|
||
def test_s7g_another_dispatch_running_is_not_this_dispatchs_started(mam_sandbox):
|
||
"""S7g (review R1): `running` set by a different dispatch's claim says
|
||
nothing about this prompt."""
|
||
res, published = _run_delivery(mam_sandbox, 0, status="running")
|
||
assert published == ["error"], (published, res.stderr)
|
||
|
||
|
||
def test_s7h_the_line_is_parsed_not_pattern_matched(mam_sandbox):
|
||
"""S7h (review N-4), the one SYNTHETIC line in this file, on purpose: the
|
||
parser must not depend on key order. Every other started line here comes
|
||
from the real producer (S7b/S7c/S7e/S8, S7i-S7k)."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1")
|
||
reordered = {"data": {"attempt_id": 1, "dispatch_token": "tok-1"}, "event": "started",
|
||
"job_id": "job1"}
|
||
(jobs / "job1.events.log").write_text(json.dumps(reordered) + "\n")
|
||
res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
|
||
assert "DELIVERED" in res.stdout, res.stdout + res.stderr
|
||
assert published == [], published
|
||
|
||
|
||
# ---- trial-2 regression: the real agent must be able to satisfy the ack ----
|
||
|
||
def test_s7i_the_old_create_session_command_shape_could_never_be_acknowledged(mam_sandbox):
|
||
"""S7i (trial 2): the command create_session used to hand the agent had no
|
||
--expect-dispatch-token, so the REAL producer wrote a started with no
|
||
token, and a token-checking waiter never accepted it: 180 s idle, then a
|
||
false 'unconfirmed' over a finished onboarding. Pinned as the negative it
|
||
is, so the fixture can never drift back to faking the token."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1", publish_args=[]) # old shape: no token args
|
||
line = json.loads((jobs / "job1.events.log").read_text().splitlines()[0])
|
||
assert line["event"] == "started" and "dispatch_token" not in line.get("data", {}), line
|
||
res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
|
||
assert published == ["error"], (published, res.stderr)
|
||
|
||
|
||
def test_s7j_the_shared_builders_command_is_acknowledged_by_the_real_producer(mam_sandbox):
|
||
"""S7j (trial 2, the fix): run exactly the command delegate_publish_cmd
|
||
hands the agent, through the real publish_event.py, and the waiter is
|
||
satisfied on the first probe."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1")
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
env = _producer_env(tmp_path)
|
||
env["MAM_JOBS_DIR"] = str(jobs)
|
||
env["WORKSPACE_ROOT"] = str(tmp_path)
|
||
script = ('source "%s"\n'
|
||
'cmd="$(delegate_publish_cmd job1 tok-1 1)"\n'
|
||
'echo "CMD=$cmd"\n'
|
||
'eval "$cmd --event started" >/dev/null\n'
|
||
'delegate_started_recorded job1 tok-1 && echo RECORDED\n') % lib
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=env, timeout=120)
|
||
assert "RECORDED" in res.stdout, res.stdout + res.stderr
|
||
cmd = res.stdout.split("CMD=")[1].splitlines()[0]
|
||
assert "--expect-dispatch-token tok-1" in cmd and "--attempt-id 1" in cmd, cmd
|
||
line = json.loads((jobs / "job1.events.log").read_text().splitlines()[-1])
|
||
assert line["data"]["dispatch_token"] == "tok-1", line
|
||
|
||
|
||
def test_s7k_create_session_hands_the_agent_the_builders_command(mam_sandbox):
|
||
"""S7k: the onboarding instruction is built by delegate_publish_cmd with
|
||
the job's token and attempt id - not by a hand-rolled copy that can drift."""
|
||
src = open(_create_sh(mam_sandbox)).read()
|
||
assert 'pub="$(delegate_publish_cmd "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT")"' in src
|
||
assert 'printf \'%q \' "$py_bin" "$pub_script" --registry-dir' not in src
|
||
|
||
|
||
def test_s7l_the_builder_matches_the_wrappers_pub_cmd_argument_for_argument(mam_sandbox):
|
||
"""S7l (req. 2, one contract): the wrapper's _pub_cmd and lib.sh's
|
||
delegate_publish_cmd produce the same argv for the same job."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1")
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||
env = _producer_env(tmp_path)
|
||
env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
|
||
"MAM_DELEGATE_JOB_NO_MAIN": "1"})
|
||
script = ('source "%s"\n'
|
||
'source "%s"\n'
|
||
'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; JOB_ID=job1; LOGS_DIR_ABS="$DELEGATE_JOB_LOGS_DIR"\n'
|
||
'echo "W=$(_pub_cmd --expect-dispatch-token tok-1 --attempt-id 1)"\n'
|
||
'echo "L=$(delegate_publish_cmd job1 tok-1 1)"\n') % (lib, wrapper, jobs)
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=env, timeout=120)
|
||
w = res.stdout.split("W=")[1].splitlines()[0].split()
|
||
l = res.stdout.split("L=")[1].splitlines()[0].split()
|
||
# same script, registry, job and token/attempt flags; the interpreter path
|
||
# may differ in spelling (the wrapper resolves it relative), so compare from
|
||
# the script name on
|
||
def tail(argv):
|
||
i = next(i for i, a in enumerate(argv) if a.endswith("publish_event.py"))
|
||
return argv[i:]
|
||
assert tail(w) == tail(l), (w, l)
|
||
|
||
|
||
def test_s7m_a_job_the_agent_already_finished_is_not_marked_error(mam_sandbox):
|
||
"""S7m (req. 4): with no started naming this dispatch but the job already
|
||
`completed`, create_session leaves it alone - no `error` over a terminal
|
||
state (the trial-2 'refused error ... after terminal completed')."""
|
||
res, published = _run_delivery(mam_sandbox, 0, status="completed")
|
||
assert published == [], (published, res.stderr)
|
||
assert "already 'completed'" in res.stderr, res.stderr
|
||
assert "DELIVERED" in res.stdout
|
||
|
||
|
||
def _wrapper_wait(tmp_path, jobs, disp_tok, ack_s=2, logs_dir=None):
|
||
"""Run the wrapper's _wait_started exactly as the wrapper runs it."""
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||
env = _producer_env(tmp_path)
|
||
env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
|
||
"MAM_DELEGATE_JOB_NO_MAIN": "1"})
|
||
script = ('source "%s"\nsource "%s"\nset +e\n'
|
||
'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; ACK_TIMEOUT=%d; START_TIME=$(date +%%s); TIMEOUT=60\n'
|
||
'LOGS_DIR_ABS="%s"\n'
|
||
'_disp_tok=%s; _wait_started job1 %d; echo "rc=$?"\n'
|
||
) % (lib, wrapper, jobs, ack_s, logs_dir or env["DELEGATE_JOB_LOGS_DIR"], disp_tok, ack_s)
|
||
return subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=env, timeout=120)
|
||
|
||
|
||
@pytest.mark.parametrize("terminal", ["error", "cancelled"])
|
||
def test_w1_wrapper_a_job_closed_without_a_started_is_not_acknowledged(mam_sandbox, terminal):
|
||
"""W1 (review R1, real producer, record NOT reset): the agent never
|
||
published started; someone published `%s`. The record left `pending`.
|
||
The wrapper's wait must still say no - as create's does."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1", publish_args=None)
|
||
env = _producer_env(tmp_path)
|
||
subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
|
||
"--event", terminal, "--detail", "closed by another writer",
|
||
"--expect-dispatch-token", "tok-1", "--attempt-id", "1"],
|
||
check=True, capture_output=True, text=True, env=env, timeout=60)
|
||
assert json.loads((jobs / "job1.json").read_text())["status"] == terminal
|
||
res = _wrapper_wait(tmp_path, jobs, "tok-1")
|
||
assert "rc=1" in res.stdout, res.stdout + res.stderr
|
||
assert "treating the prompt as delivered" not in res.stderr, res.stderr
|
||
|
||
|
||
def test_w2_wrapper_another_dispatchs_started_and_running_is_not_this_ones(mam_sandbox):
|
||
"""W2 (review R1, real producer, record NOT reset): started published
|
||
under token A (record -> running), then the job was re-dispatched with
|
||
token B. Waiting for B must not accept A's started, nor the `running`."""
|
||
tmp_path = mam_sandbox
|
||
jobs = tmp_path / ".mam" / "jobs"; jobs.mkdir(parents=True, exist_ok=True)
|
||
env = _producer_env(tmp_path)
|
||
subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
|
||
"--prompt", "x", "--agent", "claude-code", "--agent-session", "herdr:x",
|
||
"--dispatch-token", "tok-A", "--job-id", "job1", "--ack-timeout", "2"],
|
||
check=True, capture_output=True, text=True, env=env)
|
||
subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
|
||
"--event", "started", "--expect-dispatch-token", "tok-A", "--attempt-id", "1"],
|
||
check=True, capture_output=True, text=True, env=env, timeout=60)
|
||
rec = jobs / "job1.json"; d = json.loads(rec.read_text())
|
||
assert d["status"] == "running"
|
||
d["dispatch_token"] = "tok-B"; rec.write_text(json.dumps(d)) # the re-dispatch; status untouched
|
||
res = _wrapper_wait(tmp_path, jobs, "tok-B")
|
||
assert "rc=1" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
def test_w3_wrapper_this_dispatchs_started_and_running_is_acknowledged(mam_sandbox):
|
||
"""W3 (real producer, record NOT reset): the positive twin of W2."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1", started_token="tok-1")
|
||
assert json.loads((jobs / "job1.json").read_text())["status"] == "running"
|
||
res = _wrapper_wait(tmp_path, jobs, "tok-1")
|
||
assert "rc=0" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
def test_w4_both_waits_agree_on_every_case(mam_sandbox):
|
||
"""W4 (ack-finding req. 2): create's delegate_wait_started and the
|
||
wrapper's _wait_started answer the same on the four reviewer cases."""
|
||
tmp_path = mam_sandbox
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
cases = []
|
||
for label, status, started_tok, wait_tok in (
|
||
("error-no-started", "error", None, "tok-1"),
|
||
("cancelled-no-started", "cancelled", None, "tok-1"),
|
||
("other-dispatch-started", "running", "tok-1", "tok-B"),
|
||
("own-started", "running", "tok-1", "tok-1")):
|
||
jobs = _seed_job(tmp_path, "job1", status=status, started_token=started_tok)
|
||
env = _producer_env(tmp_path)
|
||
env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path)})
|
||
c = subprocess.run(["bash", "-c", 'source "%s"; delegate_wait_started job1 1 %s; echo "rc=$?"' % (lib, wait_tok)],
|
||
capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60)
|
||
w = _wrapper_wait(tmp_path, jobs, wait_tok, ack_s=1)
|
||
cases.append((label, c.stdout.strip().splitlines()[-1], w.stdout.strip().splitlines()[-1]))
|
||
for label, c_rc, w_rc in cases:
|
||
assert c_rc == w_rc, (label, cases)
|
||
assert [c for _, c, _ in cases] == ["rc=1", "rc=1", "rc=1", "rc=0"], cases
|
||
|
||
|
||
def test_w5_a_remote_agents_started_counts_only_through_the_subscribers_verified_record(mam_sandbox):
|
||
"""W5 (remote route): a remote agent leaves no line in events.log; the
|
||
subscriber writes an audit record after HMAC verification and after
|
||
dropping any dispatch_token mismatch, and marks it dispatch_verified.
|
||
SYNTHETIC on purpose: the record is written in exactly the shape
|
||
mqtt_common.append_event produces (a live subscriber needs a broker).
|
||
With a token demanded, only a verified record counts."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1")
|
||
logs = tmp_path / ".mam" / "delegate_job_logs"
|
||
(logs / "job1").mkdir(parents=True, exist_ok=True)
|
||
audit = logs / "job1" / "events.ndjson"
|
||
|
||
def rec(**kw):
|
||
base = {"event": "received", "source_event": "started", "seq": 1, "topic": "t",
|
||
"timestamp": "2026-09-22T00:00:00Z", "detail": "", "logged_at": "x"}
|
||
base.update(kw)
|
||
return json.dumps(base) + "\n"
|
||
|
||
import hashlib
|
||
fp = lambda t: hashlib.sha256(t.encode()).hexdigest()
|
||
audit.write_text(rec(dispatch_verified=False, dispatch_fingerprint=None)) # untokened: not attributed
|
||
assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
|
||
audit.write_text(rec(source_event="progress", dispatch_verified=True, dispatch_fingerprint=fp("tok-1")))
|
||
assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout # not a started
|
||
audit.write_text(rec(dispatch_verified=True)) # verified, no identity: fail closed
|
||
assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
|
||
audit.write_text(rec(dispatch_verified=True, dispatch_fingerprint=fp("tok-OLD"))) # another dispatch's
|
||
assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
|
||
audit.write_text(rec(dispatch_verified=True, dispatch_fingerprint=fp("tok-1")))
|
||
assert "rc=0" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
|
||
|
||
|
||
def test_w6_the_subscriber_stamps_dispatch_verified_on_what_it_records(mam_sandbox):
|
||
"""W6: the real subscriber code path - a _Watcher with an expected
|
||
dispatch token records a matching started as dispatch_verified, and drops
|
||
a mismatching one before recording anything."""
|
||
import types
|
||
sys.path.insert(0, SKILLS)
|
||
from lib_py.delegation import job_subscriber, mqtt_common
|
||
logs = mam_sandbox / ".mam" / "delegate_job_logs"
|
||
w = job_subscriber._Watcher({"job1"}, {"job1": None}, {"job1": 0}, {"job1": "tok-1"}, {"job1": 1})
|
||
msg = types.SimpleNamespace(topic="python/mqtt/jobs/job1/events")
|
||
good = {"schema_version": 1, "seq": 1, "job_id": "job1", "event": "started", "timestamp": "t",
|
||
"detail": "", "data": {"dispatch_token": "tok-1", "attempt_id": 1}}
|
||
bad = dict(good, seq=2, data={"dispatch_token": "tok-OLD", "attempt_id": 1})
|
||
for payload in (good, bad):
|
||
w._on_event(msg, payload, "job1") if hasattr(w, "_on_event") else w.on_message(None, None, types.SimpleNamespace(topic=msg.topic, payload=json.dumps(payload).encode()))
|
||
recs = list(mqtt_common.iter_logged_events("job1", str(logs)))
|
||
started = [r for r in recs if r.get("source_event") == "started"]
|
||
assert len(started) == 1 and started[0]["dispatch_verified"] is True, recs
|
||
import hashlib
|
||
assert started[0]["dispatch_fingerprint"] == hashlib.sha256(b"tok-1").hexdigest(), started[0]
|
||
assert started[0]["attempt_id"] == 1, started[0]
|
||
assert "tok-1" not in json.dumps(started[0]), "the token itself must never reach the audit log"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# review round 4 (planner-reviewer-claude-01 B-1, planner-reviewer-codex-01):
|
||
# the remote audit record must name the dispatch it vouches for. Real signed
|
||
# producer, real _Watcher.on_message, real audit writer, real re-dispatch
|
||
# (registry operator write minting a new token) - no record is edited by hand.
|
||
# --------------------------------------------------------------------------
|
||
|
||
def _remote_fixture(tmp_path, monkeypatch):
|
||
import test_issue18_liveness_contract as live
|
||
jobs = tmp_path / "jobs"
|
||
env = live._env(tmp_path, dict(live.UNREACHABLE, MAM_DELEGATE_JOB_NO_MAIN="1",
|
||
MAM_JOBS_DIR=str(jobs)))
|
||
for key in ("MAM_ENV_FILE", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT"):
|
||
monkeypatch.setenv(key, env[key])
|
||
sys.path.insert(0, str(live.SKILLS))
|
||
from lib_py.delegation import job_subscriber, mqtt_common, publish_event, registry
|
||
return live, env, jobs, job_subscriber, mqtt_common, publish_event, registry
|
||
|
||
|
||
def _receive(publish_event, watcher, jid, token, attempt, seq, secret):
|
||
"""A signed `started` from the real producer, handed to the real subscriber."""
|
||
import types
|
||
payload = publish_event.build_payload(jid, seq, "started", "",
|
||
{"dispatch_token": token, "attempt_id": attempt}, secret)
|
||
watcher.on_message(None, None, types.SimpleNamespace(
|
||
topic="python/mqtt/jobs/%s/events" % jid, payload=json.dumps(payload).encode()))
|
||
|
||
|
||
def _both_waits(live, env, tmp_path, jobs, jid, token):
|
||
"""create's delegate_wait_started and the wrapper's _wait_started, same inputs."""
|
||
import shlex
|
||
script = (
|
||
'source %s\nset +e\nPY=%s\nREGISTRY_DIR=%s\nACK_TIMEOUT=0\nTIMEOUT=60\nSTART_TIME=$(date +%%s)\n'
|
||
'LOGS_DIR_ABS=%s\n_disp_tok=%s\n'
|
||
'delegate_wait_started %s 0 "$_disp_tok"; echo CREATE_RC=$?\n'
|
||
'_wait_started %s 0; echo WRAPPER_RC=$?\n'
|
||
) % (shlex.quote(str(live.WRAPPER)), shlex.quote(sys.executable), shlex.quote(str(jobs)),
|
||
shlex.quote(env["DELEGATE_JOB_LOGS_DIR"]), shlex.quote(token), jid, jid)
|
||
res = subprocess.run(["bash", "-c", script], env=env, cwd=str(tmp_path),
|
||
capture_output=True, text=True, timeout=60)
|
||
out = res.stdout + res.stderr
|
||
c = int(out.split("CREATE_RC=")[1].split()[0]); w = int(out.split("WRAPPER_RC=")[1].split()[0])
|
||
return c, w, out
|
||
|
||
|
||
@pytest.mark.parametrize("scenario,expected", [
|
||
("current", 0), ("old_audit", 1), ("old_audit_and_stale_wire", 1),
|
||
("new_dispatch_received", 0), ("invalid_hmac", 1),
|
||
])
|
||
def test_w7_remote_audit_is_bound_to_the_dispatch_being_waited_for(tmp_path, monkeypatch, scenario, expected):
|
||
"""W7 (codex's five cases, real paths throughout). Attempt 1 under token A
|
||
leaves a verified audit record; the operator re-dispatches (attempt 2,
|
||
new token B, nothing edited by hand); waiting for B must not be answered
|
||
by A's record, even when B's subscriber correctly dropped A's replayed
|
||
packet - and B's own verified started must still be accepted. Both waits
|
||
must agree."""
|
||
live, env, jobs, job_subscriber, mqtt_common, publish_event, registry = _remote_fixture(tmp_path, monkeypatch)
|
||
jid, secret = "auditjob", "test-only-hmac-secret"
|
||
registry.register_job("remote audit", job_id=jid, registry_dir=str(jobs), auth_token=secret,
|
||
dispatch_token="dispatch-A", transport="remote",
|
||
broker={"host": "127.0.0.1", "port": 9, "tls": False})
|
||
w1 = job_subscriber._Watcher({jid}, {jid: secret}, {jid: 0}, {jid: "dispatch-A"}, {jid: 1})
|
||
_receive(publish_event, w1, jid, "dispatch-A", 1, 1, "wrong-secret" if scenario == "invalid_hmac" else secret)
|
||
recs = list(mqtt_common.iter_logged_events(jid, env["DELEGATE_JOB_LOGS_DIR"]))
|
||
assert sum(r.get("source_event") == "started" for r in recs) == (0 if scenario == "invalid_hmac" else 1), recs
|
||
token = "dispatch-A"
|
||
if scenario not in ("current", "invalid_hmac"):
|
||
job_subscriber._sync_remote_status(jid, "completed", str(jobs), token)
|
||
assert registry._operator_write(jid, str(jobs), status="pending", attempt_id=2, iteration=2) == 0
|
||
record = mqtt_common.load_job(jid, str(jobs))
|
||
token = record["dispatch_token"]
|
||
assert token != "dispatch-A" and record["attempt_id"] == 2, record
|
||
w2 = job_subscriber._Watcher({jid}, {jid: secret}, {jid: 0}, {jid: token}, {jid: 2})
|
||
if scenario == "old_audit_and_stale_wire":
|
||
_receive(publish_event, w2, jid, "dispatch-A", 1, 2, secret) # A's packet replayed to B's subscriber
|
||
assert w2.events.empty(), "the subscriber must drop the stale dispatch"
|
||
elif scenario == "new_dispatch_received":
|
||
_receive(publish_event, w2, jid, token, 2, 2, secret)
|
||
assert not w2.events.empty()
|
||
assert not (jobs / ("%s.events.log" % jid)).exists(), "remote evidence comes only from the subscriber's audit"
|
||
c, w, out = _both_waits(live, env, tmp_path, jobs, jid, token)
|
||
assert (c, w) == (expected, expected), (scenario, c, w, out)
|
||
|
||
|
||
def test_w8_a_record_without_identity_fails_closed_when_a_token_is_demanded(tmp_path, monkeypatch):
|
||
"""W8: an audit record from before dispatch_fingerprint existed (verified,
|
||
no fingerprint) names no dispatch: rejected when a token is demanded,
|
||
accepted when none is (an untokened legacy job), exactly as before."""
|
||
live, env, jobs, job_subscriber, mqtt_common, publish_event, registry = _remote_fixture(tmp_path, monkeypatch)
|
||
jid = "legacyjob"
|
||
registry.register_job("legacy", job_id=jid, registry_dir=str(jobs), auth_token=None,
|
||
dispatch_token="dispatch-L", transport="remote",
|
||
broker={"host": "127.0.0.1", "port": 9, "tls": False})
|
||
mqtt_common.append_event(jid, {"event": "received", "source_event": "started", "seq": 1,
|
||
"topic": "t", "timestamp": "t", "detail": "",
|
||
"dispatch_verified": True}, env["DELEGATE_JOB_LOGS_DIR"])
|
||
c, w, out = _both_waits(live, env, tmp_path, jobs, jid, "dispatch-L")
|
||
assert (c, w) == (1, 1), ("token demanded: a record with no identity must not count", out)
|
||
c, w, out = _both_waits(live, env, tmp_path, jobs, jid, "")
|
||
assert (c, w) == (0, 0), ("no token demanded: compatibility path", out)
|
||
|
||
|
||
def test_s7n_the_wrappers_wait_started_accepts_the_real_producers_line(mam_sandbox):
|
||
"""S7n (req. 2): the delegate-job wrapper's _wait_started, on the same
|
||
real-producer started (token in data), accepts it; on a stale token it
|
||
does not - via the shared predicate."""
|
||
tmp_path = mam_sandbox
|
||
jobs = _seed_job(tmp_path, "job1", started_token="tok-1")
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
|
||
env = _producer_env(tmp_path)
|
||
env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
|
||
"MAM_DELEGATE_JOB_NO_MAIN": "1"})
|
||
# the record stays exactly as the real producer left it (`running`):
|
||
# review R1 found the previous version resetting it to `pending`, which
|
||
# hid the wrapper's status bypass behind the predicate
|
||
script = ('source "%s"\nsource "%s"\n'
|
||
'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; ACK_TIMEOUT=2; START_TIME=$(date +%%s); TIMEOUT=60\n'
|
||
'_disp_tok=tok-1; _wait_started job1 2 && echo ACCEPT_OWN\n'
|
||
'_disp_tok=tok-X; _wait_started job1 2 || echo REJECT_STALE\n') % (lib, wrapper, jobs)
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=env, timeout=120)
|
||
assert "ACCEPT_OWN" in res.stdout, res.stdout + res.stderr
|
||
assert "REJECT_STALE" in res.stdout, res.stdout + res.stderr
|
||
|
||
|
||
def test_s7d_no_started_marks_the_job_but_keeps_the_session(mam_sandbox):
|
||
"""S7d: the job failed, the session did not. The handshake is marked
|
||
`error` and warned about loudly, but a live agent is not killed over a
|
||
slow onboarding - the same rule this script already applies one step
|
||
earlier to a TUI that never signalled ready."""
|
||
res, published = _run_delivery(mam_sandbox, 0)
|
||
assert res.returncode == 0, res.stdout + res.stderr
|
||
assert published == ["error"], published
|
||
assert "did not acknowledge the onboarding prompt" in res.stderr
|
||
assert "preserved" in res.stderr
|
||
|
||
|
||
def test_s7e_a_started_from_another_dispatch_is_not_evidence(mam_sandbox):
|
||
"""S7e: a loop job keeps one events.log across attempts, so only a started
|
||
carrying THIS dispatch token counts. The stale line is made by the real
|
||
producer under the record's *previous* token, then the record is
|
||
re-dispatched (token rotated) before the waiter looks."""
|
||
tmp_path = mam_sandbox
|
||
jobs = tmp_path / ".mam" / "jobs"
|
||
jobs.mkdir(parents=True, exist_ok=True)
|
||
env = _producer_env(tmp_path)
|
||
subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
|
||
"--prompt", "x", "--agent", "claude-code", "--agent-session", "herdr:x",
|
||
"--dispatch-token", "tok-OLD", "--job-id", "job1", "--ack-timeout", "2"],
|
||
check=True, capture_output=True, text=True, env=env)
|
||
subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
|
||
"--event", "started", "--expect-dispatch-token", "tok-OLD", "--attempt-id", "1"],
|
||
check=True, capture_output=True, text=True, env=env, timeout=60)
|
||
rec = jobs / "job1.json"
|
||
d = json.loads(rec.read_text()); d["dispatch_token"] = "tok-1" # the re-dispatch; status untouched
|
||
rec.write_text(json.dumps(d))
|
||
res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
|
||
assert published == ["error"], (published, res.stderr)
|
||
|
||
|
||
def test_s8_rc4_is_a_warning_and_the_started_wait_decides(mam_sandbox):
|
||
"""S8 (stubbed injection): rc 4 is "Enter not seen accepted", which is
|
||
exactly where a still-starting TUI lands - #23's false failure. It must no
|
||
longer abort; the started wait decides both ways."""
|
||
res, published = _run_delivery(mam_sandbox, 4, started_token="tok-1")
|
||
assert "DELIVERED" in res.stdout, res.stdout + res.stderr
|
||
assert "rc=4" in res.stderr, res.stderr
|
||
assert published == [], published
|
||
|
||
res, published = _run_delivery(mam_sandbox, 4)
|
||
assert published == ["error"], published
|
||
assert "did not acknowledge the onboarding prompt" in res.stderr
|
||
|
||
|
||
@pytest.mark.parametrize("rc", [1, 2, 3])
|
||
def test_s8b_the_other_injection_failures_still_abort_at_once(mam_sandbox, rc):
|
||
"""S8b: rc 1/2/3 say the pane never quiesced, a dialog was blocking, or the
|
||
text never appeared. No later `started` repairs those, so they stay hard
|
||
failures - only rc 4 was relaxed."""
|
||
res, published = _run_delivery(mam_sandbox, rc, started_token="tok-1")
|
||
assert res.returncode == 1, res.stdout + res.stderr
|
||
assert "DELIVERED" not in res.stdout
|
||
assert published == ["error"], published
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# item 4 — a rollback verifies the kill (#23)
|
||
# --------------------------------------------------------------------------
|
||
|
||
def _run_rollback(tmp_path, kill_rc, still_alive, row=True, job_id="job1"):
|
||
lib = tmp_path / ".agents" / "skills" / "lib.sh"
|
||
yaml_path = tmp_path / ".mam" / "agent-sessions.yaml"
|
||
if row:
|
||
yaml_path.write_text(
|
||
"herdr_sessions:\n"
|
||
"- name: test-sess\n"
|
||
" status: running\n"
|
||
" role: creator\n"
|
||
" delegate_job_id: %s\n"
|
||
" pane:\n"
|
||
" index: 0\n"
|
||
" pid: 111\n"
|
||
" cmd: claude\n"
|
||
" cwd: %s\n" % (job_id, tmp_path))
|
||
else:
|
||
yaml_path.write_text("herdr_sessions: []\n")
|
||
_seed_job(tmp_path, job_id)
|
||
block = _extract(_create_sh(tmp_path), "# _mark_row_rolled_back <status> <note>",
|
||
"trap cleanup_herdr_on_error EXIT")
|
||
pub_log = tmp_path / "published.txt"
|
||
# The trap is installed and fired the way create_session fires it: an
|
||
# errexit shell exiting non-zero. rc 9 through means "no exit 43".
|
||
script = '''
|
||
set -euo pipefail
|
||
source "%s"
|
||
SESSION_NAME=test-sess
|
||
DELEGATE_JOB_ID=%s
|
||
DELEGATE_JOB_TOKEN=tok-1
|
||
DELEGATE_JOB_ATTEMPT=1
|
||
ROW_APPENDED=%d
|
||
AGENT_SESSIONS_YAML="%s"
|
||
delegate_publish_event() { printf '%%s\\n' "$2" >> "%s"; }
|
||
_herdr() {
|
||
case "${1:-}" in
|
||
kill-session) return %d ;;
|
||
has-session) return %d ;;
|
||
esac
|
||
return 0
|
||
}
|
||
%s
|
||
exit 9
|
||
''' % (lib, job_id, 1 if row else 0, yaml_path, pub_log,
|
||
kill_rc, 0 if still_alive else 1, block)
|
||
env = dict(os.environ, MAM_JOBS_DIR=str(tmp_path / ".mam" / "jobs"),
|
||
AGENT_SESSIONS_YAML=str(yaml_path))
|
||
res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
|
||
cwd=str(tmp_path), env=env)
|
||
published = pub_log.read_text().split() if pub_log.exists() else []
|
||
return res, published, yaml_path.read_text()
|
||
|
||
|
||
def test_s9_a_kill_that_left_the_session_alive_is_reported(mam_sandbox):
|
||
"""S9 (stubbed herdr, real YAML + registry): the kill failed and the
|
||
session is still up. #23's rollback swallowed exactly this and printed
|
||
"rolled back" anyway, leaving a `running` row against a failed job."""
|
||
res, published, yaml_text = _run_rollback(mam_sandbox, kill_rc=1, still_alive=True)
|
||
assert res.returncode == 43, res.stdout + res.stderr
|
||
assert "test-sess" in res.stderr and "kill-session" in res.stderr
|
||
assert "rollback-failed" in yaml_text, yaml_text
|
||
assert "status: running" in yaml_text, "a session that is still up must not be called gone"
|
||
assert "delegate_job_id: null" in yaml_text, yaml_text
|
||
assert published == ["cancelled"], published
|
||
|
||
|
||
def test_s10_a_successful_kill_keeps_the_previous_behaviour(mam_sandbox):
|
||
"""S10: nothing is still running, so no exit 43 and no warning - but the
|
||
row and the job are still cleaned up, which is the other half of #23."""
|
||
res, published, yaml_text = _run_rollback(mam_sandbox, kill_rc=0, still_alive=False)
|
||
assert res.returncode == 9, res.stdout + res.stderr
|
||
assert "rolled-back" in yaml_text, yaml_text + "\nSTDERR:" + res.stderr
|
||
assert "status: terminated" in yaml_text, yaml_text
|
||
assert "delegate_job_id: null" in yaml_text, yaml_text
|
||
assert published == ["cancelled"], published
|
||
|
||
|
||
def test_s10b_a_rollback_before_the_row_existed_touches_no_yaml(mam_sandbox):
|
||
"""S10b: failing before the append must not invent a row."""
|
||
res, published, yaml_text = _run_rollback(mam_sandbox, kill_rc=0, still_alive=False,
|
||
row=False)
|
||
assert res.returncode == 9, res.stdout + res.stderr
|
||
assert "herdr_sessions: []" in yaml_text, yaml_text
|