.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 <job_id> [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 <job_id> [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 <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 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 <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
     fi
     if [ "$(date +%s)" -ge "$deadline" ]; then

.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

.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

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):