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 <noreply@anthropic.com>
This commit is contained in:
1 parent
f7cee00dec
commit
1667fda8dd
4 files changed
+303
-86
No files matched your search
+78
-32
@@ -2173,37 +2173,66 @@ delegate_publish_event() {
|
|||||||
_delegation_py publish --job "$job_id" --event "$event" --detail "$detail" ${extra[@]+"${extra[@]}"} || true
|
_delegation_py publish --job "$job_id" --event "$event" --detail "$detail" ${extra[@]+"${extra[@]}"} || true
|
||||||
}
|
}
|
||||||
|
|
||||||
# delegate_wait_started <job_id> [ack_timeout_s] [dispatch_token]
|
# delegate_publish_cmd <job_id> [dispatch_token] [attempt_id]
|
||||||
#
|
# The publish command an agent is told to run, shell-quoted, with absolute
|
||||||
# ISSUE-23 / ISSUE-18: delivery is confirmed by the protocol, not by the
|
# script/registry/audit-log paths (Issue #21: an agent whose cwd is a worktree
|
||||||
# screen. `agent prompt --wait --until working` failing two seconds after the
|
# runs it as written). ONE builder for create_session's onboarding and the
|
||||||
# keys were handed over says nothing about a TUI that is still starting up -
|
# delegate-job wrapper's dispatch, so both hand the agent the same shape:
|
||||||
# that is exactly what made create_session --onboard call a healthy agent a
|
# `--expect-dispatch-token T --attempt-id A` make publish_event write
|
||||||
# failure. The evidence is the agent's own `started`: the registry record
|
# data.dispatch_token / data.attempt_id into the agent's own events, which is
|
||||||
# leaves `pending` (publish_event syncs it on this host) or a matching started
|
# the only way a `started` can name the dispatch it answers (the trial-2
|
||||||
# line lands in the job's events.log, which publish_event writes before its
|
# finding: create_session's hand-rolled command omitted them, the agent's
|
||||||
# wire publish so a slow broker cannot delay the verdict.
|
# `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
|
||||||
# Returns 0 on that evidence, 1 if it did not appear within T. Never
|
# the record, so a stale dispatch's event is refused at the source.
|
||||||
# re-injects: the single-injection contract is fail-closed.
|
delegate_publish_cmd() {
|
||||||
#
|
local job_id="$1" tok="${2:-}" att="${3:-}"
|
||||||
# The evidence is a `started` line in the job's events.log carrying THIS
|
local py_bin pub_script pub_reg pub_logs
|
||||||
# dispatch's token (nested under `data`, as publish_event writes it; the line
|
py_bin="$(_delegate_py_bin)"
|
||||||
# is parsed as JSON, not pattern-matched, so key order is irrelevant). The
|
pub_script="$WORKSPACE_ROOT/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py"
|
||||||
# registry's status is deliberately NOT evidence: a record can leave `pending`
|
[ -f "$pub_script" ] || pub_script="$(_delegate_script publish_event.py)"
|
||||||
# because another writer closed it (`error`, `cancelled`) or because a
|
pub_script="$(cd -P "$(dirname "$pub_script")" && pwd -P)/$(basename "$pub_script")"
|
||||||
# different dispatch is `running`, none of which says this agent saw this
|
pub_reg="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:+$WORKSPACE_ROOT/.mam/jobs}}"
|
||||||
# prompt (review R1). The local publisher writes the log line before it syncs
|
pub_reg="${pub_reg:-.mam/jobs}"
|
||||||
# the status, so nothing that really started is missed.
|
mkdir -p "$pub_reg" && pub_reg="$(cd -P "$pub_reg" && pwd -P)"
|
||||||
delegate_wait_started() {
|
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 job_id="$1" ack_s="${2:-180}" tok="${3:-}"
|
local -a argv=("$py_bin" "$pub_script" --registry-dir "$pub_reg" --job "$job_id")
|
||||||
[ -n "$job_id" ] || return 1
|
[ -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 <job_id> — 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 <job_id> [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 <dispatch_token> 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 reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}"
|
||||||
local ev_log="$reg_dir/$job_id.events.log"
|
local ev_log="$reg_dir/$job_id.events.log"
|
||||||
local deadline
|
[ -f "$ev_log" ] || return 1
|
||||||
deadline=$(( $(date +%s) + ack_s ))
|
EV_LOG="$ev_log" EV_TOK="$tok" python3 -c '
|
||||||
while :; do
|
|
||||||
if [ -f "$ev_log" ] && EV_LOG="$ev_log" EV_TOK="$tok" python3 -c '
|
|
||||||
import json, os, sys
|
import json, os, sys
|
||||||
tok = os.environ.get("EV_TOK", "")
|
tok = os.environ.get("EV_TOK", "")
|
||||||
with open(os.environ["EV_LOG"], encoding="utf-8", errors="replace") as f:
|
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":
|
if not isinstance(e, dict) or e.get("event") != "started":
|
||||||
continue
|
continue
|
||||||
data = e.get("data") if isinstance(e.get("data"), dict) else {}
|
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 data.get("dispatch_token") == tok:
|
||||||
if not tok or seen == tok:
|
|
||||||
sys.exit(0)
|
sys.exit(0)
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
' 2>/dev/null; then
|
' 2>/dev/null
|
||||||
|
}
|
||||||
|
|
||||||
|
# delegate_wait_started <job_id> [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
|
return 0
|
||||||
fi
|
fi
|
||||||
if [ "$(date +%s)" -ge "$deadline" ]; then
|
if [ "$(date +%s)" -ge "$deadline" ]; then
|
||||||
|
|||||||
@@ -614,22 +614,12 @@ echo "herdr session: $SESSION_NAME (pane pid $PANE_PID, cmd $PANE_CMD, cwd $PANE
|
|||||||
if [ -n "$DELEGATE_JOB_ID" ]; then
|
if [ -n "$DELEGATE_JOB_ID" ]; then
|
||||||
echo "delegate job: $DELEGATE_JOB_ID"
|
echo "delegate job: $DELEGATE_JOB_ID"
|
||||||
|
|
||||||
# Construct instructions matching multi-agent-mux-delegate-job format
|
# The publish command the agent will run: the same shape the delegate-job
|
||||||
py_bin="$(_delegate_py_bin)"
|
# wrapper hands its agents, from the one builder in lib.sh. The dispatch
|
||||||
# Issue #21: absolute script, registry and audit-log paths (shell-quoted), so
|
# token and attempt id are what make the agent's own `started` name THIS
|
||||||
# an agent whose cwd is a git worktree can run the command as written.
|
# dispatch (trial-2 finding: without them the token-checking ack below could
|
||||||
pub_script="$WORKSPACE_ROOT/.agents/skills/multi-agent-mux-delegate-job/scripts/publish_event.py"
|
# never be satisfied by a real agent, and every onboarding sat out 180 s).
|
||||||
[ -f "$pub_script" ] || pub_script="$(_delegate_script publish_event.py)"
|
pub="$(delegate_publish_cmd "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT")"
|
||||||
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% }"
|
|
||||||
instructions="Your job_id is \"$DELEGATE_JOB_ID\" (the one just registered for THIS delegation).
|
instructions="Your job_id is \"$DELEGATE_JOB_ID\" (the one just registered for THIS delegation).
|
||||||
|
|
||||||
On start run: $pub --event started.
|
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.
|
# 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,
|
# MULTI_AGENT_RULES §6 has the GM gate on this job's `completed` anyway,
|
||||||
# so an unacknowledged onboarding already blocks real work.
|
# 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
|
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
|
fi
|
||||||
# Trigger immediate priority reconcile cycle asynchronously
|
# Trigger immediate priority reconcile cycle asynchronously
|
||||||
# Issue #21 phase 2 (A8): the control root's reconcile/watchdog, never the
|
# Issue #21 phase 2 (A8): the control root's reconcile/watchdog, never the
|
||||||
|
|||||||
@@ -646,11 +646,10 @@ _wait_started() {
|
|||||||
# The ack wait never outlives this dispatch's wall clock (START_TIME + TIMEOUT).
|
# The ack wait never outlives this dispatch's wall clock (START_TIME + TIMEOUT).
|
||||||
wall=$(( ${START_TIME:-$(date +%s)} + ${TIMEOUT:-3600} ))
|
wall=$(( ${START_TIME:-$(date +%s)} + ${TIMEOUT:-3600} ))
|
||||||
[ "$deadline" -gt "$wall" ] && deadline=$wall
|
[ "$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
|
# A loop job keeps one events.log across iterations: only a `started` that
|
||||||
# carries THIS attempt's dispatch token counts (review R1).
|
# carries THIS attempt's dispatch token counts (review R1). The predicate is
|
||||||
local started_re='"event": *"started"'
|
# lib.sh's delegate_started_recorded - the same one create_session's
|
||||||
[ -n "${_disp_tok:-}" ] && started_re='"event": *"started".*"dispatch_token": *"'"$_disp_tok"'"'
|
# onboarding ack uses, so the two waits share one definition of evidence.
|
||||||
local budget=$(( deadline - $(date +%s) ))
|
local budget=$(( deadline - $(date +%s) ))
|
||||||
while :; do
|
while :; do
|
||||||
st="$(_disk_status "$job_id")"
|
st="$(_disk_status "$job_id")"
|
||||||
@@ -660,7 +659,7 @@ _wait_started() {
|
|||||||
fi
|
fi
|
||||||
# publish_event records `started` here before its wire publish and before
|
# publish_event records `started` here before its wire publish and before
|
||||||
# the status sync, so a slow broker cannot delay the acknowledgement.
|
# 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
|
echo "inject: job $job_id recorded 'started' in events.log; treating the prompt as delivered and continuing." >&2
|
||||||
return 0
|
return 0
|
||||||
fi
|
fi
|
||||||
|
|||||||
@@ -615,19 +615,64 @@ def _extract(path, start_marker, end_marker, close=None):
|
|||||||
return "".join(lines[start:end + 1])
|
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 = tmp_path / ".mam" / "jobs"
|
||||||
jobs.mkdir(parents=True, exist_ok=True)
|
jobs.mkdir(parents=True, exist_ok=True)
|
||||||
(jobs / ("%s.json" % job_id)).write_text(json.dumps(
|
# a seed is a fresh record: the real registry refuses to re-register an
|
||||||
{"job_id": job_id, "status": status, "dispatch_token": "tok-1",
|
# existing id, and several cases seed the same id more than once
|
||||||
"attempt_id": 1, "ack_timeout_sec": 2}))
|
for stale in (jobs / ("%s.json" % job_id), jobs / ("%s.events.log" % job_id)):
|
||||||
ev = jobs / ("%s.events.log" % job_id)
|
if stale.exists():
|
||||||
if started_token is None:
|
stale.unlink()
|
||||||
if ev.exists():
|
env = _producer_env(tmp_path)
|
||||||
ev.unlink()
|
reg = subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
|
||||||
else:
|
"--prompt", "onboarding", "--agent", "claude-code",
|
||||||
ev.write_text(json.dumps(
|
"--agent-session", "herdr:x", "--dispatch-token", "tok-1",
|
||||||
{"job": job_id, "event": "started", "dispatch_token": started_token}) + "\n")
|
"--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
|
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"])
|
@pytest.mark.parametrize("status", ["error", "cancelled"])
|
||||||
def test_s7f_a_job_closed_by_someone_else_is_not_an_acknowledgement(mam_sandbox, status):
|
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
|
"""S7f (review R1 + req. 4): a record can leave `pending` because another
|
||||||
closed it. No started line from this agent means no delivery."""
|
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)
|
res, published = _run_delivery(mam_sandbox, 0, status=status)
|
||||||
assert published == ["error"], (status, published, res.stderr)
|
assert published == [], (status, published, res.stderr)
|
||||||
assert "did not acknowledge the onboarding prompt" in 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):
|
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)
|
assert published == ["error"], (published, res.stderr)
|
||||||
|
|
||||||
|
|
||||||
def test_s7h_the_real_events_log_shape_is_read(mam_sandbox):
|
def test_s7h_the_line_is_parsed_not_pattern_matched(mam_sandbox):
|
||||||
"""S7h (review N-4): publish_event nests dispatch_token under `data` and
|
"""S7h (review N-4), the one SYNTHETIC line in this file, on purpose: the
|
||||||
the line is JSON, so it is parsed, not pattern-matched. A line with the
|
parser must not depend on key order. Every other started line here comes
|
||||||
keys in another order, and a token in either place, both count."""
|
from the real producer (S7b/S7c/S7e/S8, S7i-S7k)."""
|
||||||
tmp_path = mam_sandbox
|
tmp_path = mam_sandbox
|
||||||
jobs = _seed_job(tmp_path, "job1")
|
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",
|
reordered = {"data": {"attempt_id": 1, "dispatch_token": "tok-1"}, "event": "started",
|
||||||
"job_id": "job1"}
|
"job_id": "job1"}
|
||||||
for line in (real, reordered):
|
(jobs / "job1.events.log").write_text(json.dumps(reordered) + "\n")
|
||||||
(jobs / "job1.events.log").write_text(json.dumps(line) + "\n")
|
res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
|
||||||
res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
|
assert "DELIVERED" in res.stdout, res.stdout + res.stderr
|
||||||
assert "DELIVERED" in res.stdout, (line, res.stdout + res.stderr)
|
assert published == [], published
|
||||||
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):
|
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):
|
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
|
"""S7e: a loop job keeps one events.log across attempts, so only a started
|
||||||
carrying THIS dispatch token counts."""
|
carrying THIS dispatch token counts. The stale line is made by the real
|
||||||
res, published = _run_delivery(mam_sandbox, 0, started_token="tok-OTHER")
|
producer under the record's *previous* token, then the record is
|
||||||
assert published == ["error"], published
|
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):
|
def test_s8_rc4_is_a_warning_and_the_started_wait_decides(mam_sandbox):
|
||||||
|
|||||||
Reference in new issue
Block a user