From 1667fda8dd64e1ba89f57059bc62e57718435f09 Mon Sep 17 00:00:00 2001 From: Godopu Date: Tue, 22 Sep 2026 14:42:38 +0000 Subject: [PATCH] fix(issue-23): the onboarding ack is satisfied by what a real agent publishes; one command builder and one started predicate for create and the wrapper Trial 2 of this branch: cross-server delivery was restored (f7cee00), the agent completed its onboarding and published started and completed - and create_session still waited 180 s, called the onboarding unconfirmed, and tried to write `error` over the completed job (refused by the registry). delegate_wait_started required the started line to carry data.dispatch_token (review R1: a stale attempt's started must not vouch for this one). But publish_event only writes that field when its command line carries --expect-dispatch-token, and create_session's hand-rolled onboarding command never did - the wrapper's _pub_cmd does, which is why every wrapper-dispatched agent's started has the token and no create_session onboarding ever could. Five reviewers missed it because the fixture wrote the token into the started line by hand: a fake producer that differs from the real one. Fix, along the wrapper's existing mechanism rather than a new one: * delegate_publish_cmd (lib.sh) builds the publish command for an agent, with --expect-dispatch-token and --attempt-id, and create_session uses it. The agent's own started now names the dispatch, and publish_event's CAS check refuses a stale dispatch's event at the source. S7l pins the builder argument-for-argument against the wrapper's _pub_cmd. * delegate_started_recorded (lib.sh) is the one definition of "this dispatch's started is on disk" - JSON-parsed, token-matched when a token is given - used by delegate_wait_started and by the wrapper's _wait_started in place of its regex. The wrapper keeps its status and subscriber routes (remote agents); the shared part is the evidence contract. * An already-terminal job is left alone when the wait expires (req. 4): no error over a completed/error/cancelled record. The fixture now registers the record through registry.py and publishes started through the real publish_event.py; the only synthetic line left (S7h) exists to pin key-order independence and says so. S7i pins the trial-2 defect as the negative it is (old command shape -> started without token -> not acknowledged); S7j runs the builder's command through the real producer and is acknowledged on the first probe; S7n shows the wrapper accepting the same real line and rejecting a stale token; S7m pins no-write-over-terminal. S7f moves from "publish error" to "leave the terminal job alone", per req. 4. Option B (verify data.hmac_sig with the job's auth_token, plus record-token still equal) was weighed and not taken: verify_hmac returns True with no auth_token - PoC mode, which the trial and every sandbox run in - so it proves nothing there; in production hmac_sig authenticates the signer but does not identify the dispatch, so R1 would rest on a current-state check that a stale line plus an unrotated token passes; and it would diverge from the wrapper unless the wrapper changed too. Token exposure is not new: every wrapper- dispatched instruction already carries the dispatch token in the pane, and it is a per-dispatch CAS nonce, not the HMAC secret. Co-Authored-By: Claude Fable 5.1 --- .agents/skills/lib.sh | 110 ++++++--- .../scripts/create_session.sh | 40 +-- .../multi-agent-mux-delegate-job | 9 +- tests/test_issue23_24_startup_path.py | 230 +++++++++++++++--- 4 files changed, 303 insertions(+), 86 deletions(-) diff --git a/.agents/skills/lib.sh b/.agents/skills/lib.sh index cc3e4b4..8f5e99d 100644 --- a/.agents/skills/lib.sh +++ b/.agents/skills/lib.sh @@ -2173,37 +2173,66 @@ delegate_publish_event() { _delegation_py publish --job "$job_id" --event "$event" --detail "$detail" ${extra[@]+"${extra[@]}"} || true } -# delegate_wait_started [ack_timeout_s] [dispatch_token] -# -# ISSUE-23 / ISSUE-18: delivery is confirmed by the protocol, not by the -# screen. `agent prompt --wait --until working` failing two seconds after the -# keys were handed over says nothing about a TUI that is still starting up - -# that is exactly what made create_session --onboard call a healthy agent a -# failure. The evidence is the agent's own `started`: the registry record -# leaves `pending` (publish_event syncs it on this host) or a matching started -# line lands in the job's events.log, which publish_event writes before its -# wire publish so a slow broker cannot delay the verdict. -# -# Returns 0 on that evidence, 1 if it did not appear within T. Never -# re-injects: the single-injection contract is fail-closed. -# -# The evidence is a `started` line in the job's events.log carrying THIS -# dispatch's token (nested under `data`, as publish_event writes it; the line -# is parsed as JSON, not pattern-matched, so key order is irrelevant). The -# registry's status is deliberately NOT evidence: a record can leave `pending` -# because another writer closed it (`error`, `cancelled`) or because a -# different dispatch is `running`, none of which says this agent saw this -# prompt (review R1). The local publisher writes the log line before it syncs -# the status, so nothing that really started is missed. -delegate_wait_started() { - local job_id="$1" ack_s="${2:-180}" tok="${3:-}" - [ -n "$job_id" ] || return 1 +# delegate_publish_cmd [dispatch_token] [attempt_id] +# The publish command an agent is told to run, shell-quoted, with absolute +# script/registry/audit-log paths (Issue #21: an agent whose cwd is a worktree +# runs it as written). ONE builder for create_session's onboarding and the +# delegate-job wrapper's dispatch, so both hand the agent the same shape: +# `--expect-dispatch-token T --attempt-id A` make publish_event write +# data.dispatch_token / data.attempt_id into the agent's own events, which is +# the only way a `started` can name the dispatch it answers (the trial-2 +# finding: create_session's hand-rolled command omitted them, the agent's +# `started` carried no token, and the token-checking waiter could never be +# satisfied by a real agent). publish_event also CAS-checks that token against +# the record, so a stale dispatch's event is refused at the source. +delegate_publish_cmd() { + local job_id="$1" tok="${2:-}" att="${3:-}" + local py_bin pub_script pub_reg pub_logs + py_bin="$(_delegate_py_bin)" + pub_script="$WORKSPACE_ROOT/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py" + [ -f "$pub_script" ] || pub_script="$(_delegate_script publish_event.py)" + pub_script="$(cd -P "$(dirname "$pub_script")" && pwd -P)/$(basename "$pub_script")" + pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}" + pub_reg="${pub_reg:-.mam/jobs}" + mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)" + pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' 2>/dev/null || true)" + local -a argv=("$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$job_id") + [ -n "$tok" ] && argv+=(--expect-dispatch-token "$tok") + [ -n "$att" ] && argv+=(--attempt-id "$att") + local out + out="$( + if [ -n "$pub_logs" ]; then printf 'DELEGATE_JOB_LOGS_DIR=%q ' "$pub_logs"; fi + printf '%q ' "${argv[@]}" + )" + printf '%s' "${out% }" +} + +# delegate_job_status — the record's status, empty if unreadable. +delegate_job_status() { + local reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}" + REC="$reg_dir/$1.json" python3 -c ' +import json, os +try: + print(json.load(open(os.environ["REC"])).get("status", "") or "") +except Exception: + print("") +' 2>/dev/null || true +} + +# delegate_started_recorded [dispatch_token] +# THE definition of "this dispatch's started is on disk", shared by +# delegate_wait_started (create_session's onboarding ack) and the delegate-job +# wrapper's _wait_started: a `started` line in the job's events.log whose +# data.dispatch_token equals when one is given (a loop job +# keeps one events.log across attempts, and a stale attempt's started must not +# vouch for this one - review R1). The line is parsed as JSON, never +# pattern-matched. rc 0 = recorded, 1 = not (yet). +delegate_started_recorded() { + local job_id="$1" tok="${2:-}" local reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}" local ev_log="$reg_dir/$job_id.events.log" - local deadline - deadline=$(( $(date +%s) + ack_s )) - while :; do - if [ -f "$ev_log" ] && EV_LOG="$ev_log" EV_TOK="$tok" python3 -c ' + [ -f "$ev_log" ] || return 1 + EV_LOG="$ev_log" EV_TOK="$tok" python3 -c ' import json, os, sys tok = os.environ.get("EV_TOK", "") with open(os.environ["EV_LOG"], encoding="utf-8", errors="replace") as f: @@ -2215,11 +2244,28 @@ with open(os.environ["EV_LOG"], encoding="utf-8", errors="replace") as f: if not isinstance(e, dict) or e.get("event") != "started": continue data = e.get("data") if isinstance(e.get("data"), dict) else {} - seen = data.get("dispatch_token") or e.get("dispatch_token") or "" - if not tok or seen == tok: + if not tok or data.get("dispatch_token") == tok: sys.exit(0) sys.exit(1) -' 2>/dev/null; then +' 2>/dev/null +} + +# delegate_wait_started [ack_timeout_s] [dispatch_token] +# +# ISSUE-23 / ISSUE-18: delivery is confirmed by the protocol, not by the +# screen. The evidence is delegate_started_recorded (above); the registry's +# status is deliberately NOT evidence - a record leaves `pending` when another +# writer closes it or another dispatch claims it (review R1). The local +# publisher writes the log line before it syncs the status, so nothing that +# really started is missed. Returns 0 on the evidence, 1 if it did not appear +# within T. Never re-injects: the single-injection contract is fail-closed. +delegate_wait_started() { + local job_id="$1" ack_s="${2:-180}" tok="${3:-}" + [ -n "$job_id" ] || return 1 + local deadline + deadline=$(( $(date +%s) + ack_s )) + while :; do + if delegate_started_recorded "$job_id" "$tok"; then return 0 fi if [ "$(date +%s)" -ge "$deadline" ]; then diff --git a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh index 1d9cbfc..79e4454 100755 --- a/.agents/skills/multi-agent-mux-create/scripts/create_session.sh +++ b/.agents/skills/multi-agent-mux-create/scripts/create_session.sh @@ -614,22 +614,12 @@ echo "herdr session: $SESSION_NAME (pane pid $PANE_PID, cmd $PANE_CMD, cwd $PANE if [ -n "$DELEGATE_JOB_ID" ]; then echo "delegate job: $DELEGATE_JOB_ID" - # Construct instructions matching multi-agent-mux-delegate-job format - py_bin="$(_delegate_py_bin)" - # Issue #21: absolute script, registry and audit-log paths (shell-quoted), so - # an agent whose cwd is a git worktree can run the command as written. - pub_script="$WORKSPACE_ROOT/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py" - [ -f "$pub_script" ] || pub_script="$(_delegate_script publish_event.py)" - pub_script="$(cd -P "$(dirname "$pub_script")" && pwd -P)/$(basename "$pub_script")" - pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}" - pub_reg="${pub_reg:-.mam/jobs}" - mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)" - pub_logs="$(PYTHONPATH="$SKILL_DIR:${PYTHONPATH:-}" "$py_bin" -c 'import os; from lib_py.delegation.mqtt_common import get_logs_dir; print(os.path.abspath(get_logs_dir()))' 2>/dev/null || true)" - pub="$( - if [ -n "$pub_logs" ]; then printf 'DELEGATE_JOB_LOGS_DIR=%q ' "$pub_logs"; fi - printf '%q ' "$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$DELEGATE_JOB_ID" - )" - pub="${pub% }" + # The publish command the agent will run: the same shape the delegate-job + # wrapper hands its agents, from the one builder in lib.sh. The dispatch + # token and attempt id are what make the agent's own `started` name THIS + # dispatch (trial-2 finding: without them the token-checking ack below could + # never be satisfied by a real agent, and every onboarding sat out 180 s). + pub="$(delegate_publish_cmd "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT")" instructions="Your job_id is \"$DELEGATE_JOB_ID\" (the one just registered for THIS delegation). On start run: $pub --event started. @@ -675,10 +665,22 @@ Task: $SUBMIT_JOB_PROMPT" # not rolled back. So: mark the job, warn loudly, keep the session. # MULTI_AGENT_RULES §6 has the GM gate on this job's `completed` anyway, # so an unacknowledged onboarding already blocks real work. - delegate_publish_event "$DELEGATE_JOB_ID" error "onboarding delivery unconfirmed: no started within ${ONBOARD_ACK_S}s" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT" + # A job that is already terminal is not marked again: `error` over a + # `completed` is refused by the registry anyway (trial-2 log: "conflicting + # error after terminal completed"), and a job the agent has finished is + # not an undelivered one - only the evidence was missing. + ONBOARD_JOB_ST="$(delegate_job_status "$DELEGATE_JOB_ID")" + case "$ONBOARD_JOB_ST" in + completed|error|cancelled) + echo "WARNING: no started event naming this dispatch for '$SESSION_NAME' within ${ONBOARD_ACK_S}s, but job $DELEGATE_JOB_ID is already '$ONBOARD_JOB_ST'; leaving it as it is." >&2 + ;; + *) + delegate_publish_event "$DELEGATE_JOB_ID" error "onboarding delivery unconfirmed: no started within ${ONBOARD_ACK_S}s" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT" + echo "WARNING: '$SESSION_NAME' did not acknowledge the onboarding prompt within ${ONBOARD_ACK_S}s." >&2 + echo " The session is preserved and job $DELEGATE_JOB_ID is marked error; re-send the onboarding with multi-agent-mux-delegate-job submit once the agent is up." >&2 + ;; + esac JOB_TERMINALIZED=1 - echo "WARNING: '$SESSION_NAME' did not acknowledge the onboarding prompt within ${ONBOARD_ACK_S}s." >&2 - echo " The session is preserved and job $DELEGATE_JOB_ID is marked error; re-send the onboarding with multi-agent-mux-delegate-job submit once the agent is up." >&2 fi # Trigger immediate priority reconcile cycle asynchronously # Issue #21 phase 2 (A8): the control root's reconcile/watchdog, never the diff --git a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job index 5347d02..2559b77 100755 --- a/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job +++ b/.agents/skills/multi-agent-mux-delegate-job/multi-agent-mux-delegate-job @@ -646,11 +646,10 @@ _wait_started() { # The ack wait never outlives this dispatch's wall clock (START_TIME + TIMEOUT). wall=$(( ${START_TIME:-$(date +%s)} + ${TIMEOUT:-3600} )) [ "$deadline" -gt "$wall" ] && deadline=$wall - local ev_log="$REGISTRY_DIR/$job_id.events.log" # A loop job keeps one events.log across iterations: only a `started` that - # carries THIS attempt's dispatch token counts (review R1). - local started_re='"event": *"started"' - [ -n "${_disp_tok:-}" ] && started_re='"event": *"started".*"dispatch_token": *"'"$_disp_tok"'"' + # carries THIS attempt's dispatch token counts (review R1). The predicate is + # lib.sh's delegate_started_recorded - the same one create_session's + # onboarding ack uses, so the two waits share one definition of evidence. local budget=$(( deadline - $(date +%s) )) while :; do st="$(_disk_status "$job_id")" @@ -660,7 +659,7 @@ _wait_started() { fi # publish_event records `started` here before its wire publish and before # the status sync, so a slow broker cannot delay the acknowledgement. - if [ -f "$ev_log" ] && grep -Eq "$started_re" "$ev_log" 2>/dev/null; then + if MAM_JOBS_DIR="$REGISTRY_DIR" delegate_started_recorded "$job_id" "${_disp_tok:-}"; then echo "inject: job $job_id recorded 'started' in events.log; treating the prompt as delivered and continuing." >&2 return 0 fi diff --git a/tests/test_issue23_24_startup_path.py b/tests/test_issue23_24_startup_path.py index ce9c999..9fb39a4 100644 --- a/tests/test_issue23_24_startup_path.py +++ b/tests/test_issue23_24_startup_path.py @@ -615,19 +615,64 @@ def _extract(path, start_marker, end_marker, close=None): return "".join(lines[start:end + 1]) -def _seed_job(tmp_path, job_id, status="pending", started_token=None): +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) - (jobs / ("%s.json" % job_id)).write_text(json.dumps( - {"job_id": job_id, "status": status, "dispatch_token": "tok-1", - "attempt_id": 1, "ack_timeout_sec": 2})) - ev = jobs / ("%s.events.log" % job_id) - if started_token is None: - if ev.exists(): - ev.unlink() - else: - ev.write_text(json.dumps( - {"job": job_id, "event": "started", "dispatch_token": started_token}) + "\n") + # 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 @@ -687,11 +732,15 @@ def test_s7c_a_started_that_was_followed_by_a_terminal_status_still_counts(mam_s @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): a record can leave `pending` because another writer - closed it. No started line from this agent means no delivery.""" + """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 == ["error"], (status, published, res.stderr) - assert "did not acknowledge the onboarding prompt" in res.stderr + 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): @@ -701,22 +750,127 @@ def test_s7g_another_dispatch_running_is_not_this_dispatchs_started(mam_sandbox) assert published == ["error"], (published, res.stderr) -def test_s7h_the_real_events_log_shape_is_read(mam_sandbox): - """S7h (review N-4): publish_event nests dispatch_token under `data` and - the line is JSON, so it is parsed, not pattern-matched. A line with the - keys in another order, and a token in either place, both count.""" +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") - real = {"schema_version": 1, "seq": 1, "job_id": "job1", "event": "started", - "timestamp": "2026-09-22T08:04:26Z", "detail": "", - "data": {"dispatch_token": "tok-1", "attempt_id": 1}} reordered = {"data": {"attempt_id": 1, "dispatch_token": "tok-1"}, "event": "started", "job_id": "job1"} - for line in (real, reordered): - (jobs / "job1.events.log").write_text(json.dumps(line) + "\n") - res, published = _run_delivery(mam_sandbox, 0, seed_events=False) - assert "DELIVERED" in res.stdout, (line, res.stdout + res.stderr) - assert published == [], published + (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 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"}) + # keep the record `pending` so only the events.log route can answer + rec = jobs / "job1.json"; d = json.loads(rec.read_text()); d["status"] = "pending"; rec.write_text(json.dumps(d)) + 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): @@ -733,9 +887,25 @@ def test_s7d_no_started_marks_the_job_but_keeps_the_session(mam_sandbox): 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.""" - res, published = _run_delivery(mam_sandbox, 0, started_token="tok-OTHER") - assert published == ["error"], published + 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"; d["status"] = "pending" + 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):