fix(issue-23): the remote audit record names the dispatch it vouches for; a record without identity fails closed

Review round 4 (planner-reviewer-claude-01 B-1, planner-reviewer-codex-01 P2, reached independently): dispatch_verified is a boolean. It says the subscriber's token check ran when the record was written, but names no dispatch, the token being masked by the audit redactor and attempt_id not written. events.ndjson is appended to across a job's attempts, so a started verified under attempt 1 satisfied every later attempt's wait - the very misattribution R1 forbids, alive on the remote path while the local path (data.dispatch_token in events.log) was already immune. codex reproduced it with the real signed producer, the real _Watcher, the real audit writer and the real operator re-dispatch; both waits answered 0 where 1 was due, even when attempt 2's subscriber had correctly dropped attempt 1's replayed packet. The subscriber now writes dispatch_fingerprint = SHA-256 of the token it verified, and attempt_id for readers. With a token demanded, the predicate accepts an audit record only if it is dispatch_verified AND its fingerprint equals SHA-256 of that token; a record with no fingerprint - written before the field existed - names nothing and is rejected on that path. Without a token, any received started counts, as before. Fingerprint over attempt_id as the identity, weighing the two proposals: the local path already compares the token, so hashing it keeps one identity axis for both sources; the token is a uuid4 nonce (registry.py:141), 128 bits, so its SHA-256 is not recoverable and it is a CAS fence rather than the HMAC secret in the first place; and attempt_id cannot tell apart a token re-minted within one attempt, where the token can. attempt_id is recorded because it is cheap and readable, not relied on. W7 pins codex's five cases end to end - current dispatch accepted, an old attempt's record rejected after a real re-dispatch, still rejected when the stale packet is replayed and dropped, the new dispatch's own started accepted, a bad HMAC never recorded - with create's wait and the wrapper's wait required to agree on each. W8 pins the fail-closed legacy record and the untokened compatibility path. W5 gains the no-fingerprint and other-dispatch cases; W6 checks the real _Watcher writes the fingerprint and never the token. S7e's status reset had no reason left and is gone. codex's own reproduction file, unmodified, passes 5/5 against this tree. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
fix(issue-23): both ack waits share one contract - the wrapper's status bypass is gone; the G9 guard removes only the docstring node
2026-09-22 17:52:51 +00:00 · 2026-09-22 16:43:00 +00:00 · 2026-09-22 15:32:40 +00:00 · 2026-09-22 14:42:38 +00:00 · 2026-09-22 14:13:38 +00:00 · 2026-09-22 10:25:52 +00:00
14 changed files with 1213 additions and 158 deletions
@@ -71,7 +71,6 @@ done
 _MAM_MODAL_TOKENS='Do you trust the files|Yes, proceed|No, exit|Allow this|browser to authenticate|Resuming the full session|Resume from summary|Try the new fullscreen renderer\?'
 _MAM_HINT_TOKENS='Use arrow keys|Esc to cancel|Press Enter to continue'
 _MAM_DIALOG_TOKENS="${_MAM_MODAL_TOKENS}|${_MAM_HINT_TOKENS}"
-_MAM_READY_TOKENS_CLAUDE='Anthropic|Assistant|Chat|Welcome|Claude Code|Opus|Sonnet|Haiku'

 # Workspace-relative defaults with environment overrides (Phase Z)
 HOME_DIR="${HOME_DIR:-$HOME}"
@@ -1385,19 +1384,62 @@ EOF
 # live session as absent: has-session, kill-session, agent prompt and the
 # delegate-job preflight all answer "no such session".
 #
+# _herdr_caller_server — the name of the server whose pane this process runs
+# in; empty when it is not in a pane at all.
+#
+# HERDR_SESSION is used when present, but herdr 0.8.2 does not export it to a
+# pane of the *default* server (the orchestrator's usual home), which is
+# exactly the pane the #23 leak comes from - the live trial of this branch
+# found every call judged "not crossing" for that reason. HERDR_SOCKET_PATH is
+# exported to every pane, and its layout names the server:
+#
+#   $HOME/.config/herdr/herdr.sock                 -> default
+#   $HOME/.config/herdr/sessions/<name>/herdr.sock -> <name>
+#
+# (the same convention lib.sh's shim already relies on to find a named
+# server's socket). Matched by shape, not by $HOME, so a redirected HOME does
+# not break it. An unrecognised layout yields "" - treated as "not in a pane",
+# which keeps today's behaviour rather than guessing.
+_herdr_caller_server() {
+  if [ -n "${HERDR_SESSION:-}" ]; then
+    printf '%s\n' "$HERDR_SESSION"
+    return 0
+  fi
+  local sock="${HERDR_SOCKET_PATH:-}"
+  case "$sock" in
+    */.config/herdr/herdr.sock)
+      printf 'default\n'
+      ;;
+    */.config/herdr/sessions/*/herdr.sock)
+      local d="${sock%/herdr.sock}"
+      printf '%s\n' "${d##*/}"
+      ;;
+    *)
+      printf '\n'
+      ;;
+  esac
+  return 0
+}
+
 # True only for a call aimed at a *named* server other than the caller's own:
 #
 #   * target unset or `default` (the lib.sh:97 fallback) is the shim's sentinel
 #     for "pass no session flag", so herdr answers on the ambient server, which
 #     for a caller inside a pane is that same caller's server - not a crossing;
-#   * no HERDR_SESSION means the caller is not in a pane at all, so an exported
-#     HERDR_WORKSPACE_ID is a deliberate scope, not a leak (H-18/H-21/H-23);
+#   * no caller server (_herdr_caller_server empty) means the caller is not in
+#     a pane at all, so an exported HERDR_WORKSPACE_ID is a deliberate scope,
+#     not a leak (H-18/H-21/H-23);
 #   * caller == target is the same server.
+#
+# A caller in the default server's pane targeting a named server IS a crossing
+# (the reported #23 scenario): `default` is a real server name on the caller's
+# side, only as a *target* is it the "ambient" sentinel.
 _herdr_call_is_cross_server() {
-  local caller="${HERDR_SESSION:-}" target="${HERDR_SESSION_NAME:-}"
+  local caller target="${HERDR_SESSION_NAME:-}"
  case "$target" in
    ""|default) return 1 ;;
  esac
+  caller="$(_herdr_caller_server)"
  if [ -n "$caller" ] && [ "$caller" != "$target" ]; then
    return 0
  fi
@@ -1413,7 +1455,7 @@ mam_herdr() {
  if _herdr_call_is_cross_server; then
    _scope_strip=(env -u HERDR_WORKSPACE_ID -u HERDR_PANE_ID -u HERDR_TAB_ID)
    if [ "${MAM_SHIM_DEBUG:-0}" = "1" ]; then
-      echo "mam_herdr: dropping caller pane scope ids (caller server '${HERDR_SESSION:-}' != target '${HERDR_SESSION_NAME:-}')" >&2
+      echo "mam_herdr: dropping caller pane scope ids (caller server '$(_herdr_caller_server)' != target '${HERDR_SESSION_NAME:-}')" >&2
    fi
  fi
  ${_scope_strip[@]+"${_scope_strip[@]}"} "$WORKSPACE_ROOT/.mam/shim/herdr" "$@"
@@ -2131,46 +2173,136 @@ delegate_publish_event() {
  _delegation_py publish --job "$job_id" --event "$event" --detail "$detail" ${extra[@]+"${extra[@]}"} || true
 }

+# 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). create_session's onboarding uses this; the delegate-job
+# wrapper still builds its own (_pub_cmd), and S7l pins the two equal
+# argument-for-argument, 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] [logs_dir]
+# 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. Two sources, one rule:
+#
+#  * <registry>/<job>.events.log - the local publisher's own line, whose
+#    data.dispatch_token must equal <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);
+#  * <logs_dir>/<job>/events.ndjson - the subscriber's audit record of a
+#    `started` it received, the only evidence a REMOTE agent (no shared disk)
+#    can leave. The subscriber verifies the HMAC and drops any event whose
+#    data.dispatch_token differs from the job's expected token before it
+#    writes this record (dispatch_verified), and writes dispatch_fingerprint
+#    = SHA-256 of that token. The file is appended to across a job's
+#    attempts, so the flag alone would let an earlier attempt's record
+#    acknowledge a later dispatch (review round 4): with <dispatch_token>
+#    given, a record counts only if it is dispatch_verified AND its
+#    fingerprint equals SHA-256(<dispatch_token>). A record with no
+#    fingerprint (written before this field existed) names no dispatch and
+#    is rejected on that path - fail closed. Without a token, any received
+#    started counts, as before.
+#
+# Never the registry status: a record leaves `pending` when another writer
+# closes it or another dispatch claims it. Lines are parsed as JSON, never
+# pattern-matched. rc 0 = recorded, 1 = not (yet).
+delegate_started_recorded() {
+  local job_id="$1" tok="${2:-}" logs_dir="${3:-}"
+  local reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}"
+  [ -n "$logs_dir" ] || logs_dir="${DELEGATE_JOB_LOGS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/delegate_job_logs}"
+  local ev_log="$reg_dir/$job_id.events.log"
+  local audit="$logs_dir/$job_id/events.ndjson"
+  [ -f "$ev_log" ] || [ -f "$audit" ] || return 1
+  EV_LOG="$ev_log" EV_AUDIT="$audit" EV_TOK="$tok" python3 -c '
+import hashlib, json, os, sys
+tok = os.environ.get("EV_TOK", "")
+fp = hashlib.sha256(tok.encode("utf-8")).hexdigest() if tok else ""
+
+def lines(path):
+    if not os.path.isfile(path):
+        return
+    with open(path, encoding="utf-8", errors="replace") as f:
+        for line in f:
+            try:
+                e = json.loads(line)
+            except Exception:
+                continue
+            if isinstance(e, dict):
+                yield e
+
+for e in lines(os.environ["EV_LOG"]):
+    if e.get("event") != "started":
+        continue
+    data = e.get("data") if isinstance(e.get("data"), dict) else {}
+    if not tok or data.get("dispatch_token") == tok:
+        sys.exit(0)
+for e in lines(os.environ["EV_AUDIT"]):
+    if e.get("event") != "received" or e.get("source_event") != "started":
+        continue
+    if not tok:
+        sys.exit(0)
+    if e.get("dispatch_verified") is True and e.get("dispatch_fingerprint") == fp:
+        sys.exit(0)
+sys.exit(1)
+' 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. `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 neither appeared within T. Never re-injects:
-# the single-injection contract is fail-closed.
+# 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 reg_dir="${MAM_JOBS_DIR:-${WORKSPACE_ROOT:-$PWD}/.mam/jobs}"
-  local rec="$reg_dir/$job_id.json"
-  local ev_log="$reg_dir/$job_id.events.log"
-  local started_re='"event": *"started"'
-  if [ -n "$tok" ]; then
-    started_re='"event": *"started".*"dispatch_token": *"'"$tok"'"'
-  fi
-  local deadline st
+  local deadline
  deadline=$(( $(date +%s) + ack_s ))
  while :; do
-    st=""
-    if [ -f "$rec" ]; then
-      st=$(REC="$rec" python3 -c "
-import json, os
-try:
-    print(json.load(open(os.environ['REC'])).get('status', '') or '')
-except Exception:
-    print('')
-" 2>/dev/null || true)
-    fi
-    if [ -n "$st" ] && [ "$st" != "pending" ]; then
-      return 0
-    fi
-    if [ -f "$ev_log" ] && grep -Eq "$started_re" "$ev_log" 2>/dev/null; then
+    if delegate_started_recorded "$job_id" "$tok"; then
      return 0
    fi
    if [ "$(date +%s)" -ge "$deadline" ]; then
@@ -2752,21 +2884,35 @@ send_keys_safe() {
 }

 # _mam_load_dialog_facts <agent>
-# Populate MAM_STARTUP_DIALOGS for <agent>, once per agent per shell. Only that
-# one fact is taken: `facts` is an eval contract (F8), and evaluating all of it
-# here would clobber ready/placeholder tokens a caller deliberately set.
-# An agent the adapters do not know keeps an empty list.
+# Populate MAM_STARTUP_DIALOGS for <agent>, once per agent per shell, and the
+# ready tokens (MAM_STRONG_READY_TOKENS / MAM_READY_TOKENS) only when the caller
+# has not set them. `facts` is an eval contract (F8); evaluating all of it here
+# would clobber tokens a caller deliberately set, so only these are taken, and
+# the tokens only fill a gap. Without them the handler could never recognise a
+# ready screen when called on its own - resume_session.sh does exactly that,
+# with no wait_for_tui_ready before it - and idled through its whole timeout on
+# every resume (review B-1). An agent the adapters do not know keeps an empty
+# list and no tokens.
 _mam_load_dialog_facts() {
-  local agent="$1" _f=""
+  local agent="$1" _f="" _d="" _s="" _r=""
  if [ "${_MAM_DIALOGS_AGENT:-}" = "$agent" ]; then
    return 0
  fi
-  _f="$("$(_delegate_py_bin)" -m lib_py.agents facts "$agent" 2>/dev/null | sed -n 's/^MAM_STARTUP_DIALOGS=//p')" || _f=""
-  if [ -n "$_f" ]; then
-    eval "MAM_STARTUP_DIALOGS=$_f"
+  _f="$("$(_delegate_py_bin)" -m lib_py.agents facts "$agent" 2>/dev/null)" || _f=""
+  _d="$(printf '%s\n' "$_f" | sed -n 's/^MAM_STARTUP_DIALOGS=//p')"
+  _s="$(printf '%s\n' "$_f" | sed -n 's/^MAM_STRONG_READY_TOKENS=//p')"
+  _r="$(printf '%s\n' "$_f" | sed -n 's/^MAM_READY_TOKENS=//p')"
+  if [ -n "$_d" ]; then
+    eval "MAM_STARTUP_DIALOGS=$_d"
  else
    MAM_STARTUP_DIALOGS='[]'
  fi
+  if [ -z "${MAM_STRONG_READY_TOKENS:-}" ] && [ -n "$_s" ]; then
+    eval "MAM_STRONG_READY_TOKENS=$_s"
+  fi
+  if [ -z "${MAM_READY_TOKENS:-}" ] && [ -n "$_r" ]; then
+    eval "MAM_READY_TOKENS=$_r"
+  fi
  _MAM_DIALOGS_AGENT="$agent"
  return 0
 }
@@ -24,6 +24,7 @@ Exit codes:
 from __future__ import annotations

 import argparse
+import hashlib
 import json
 import logging
 import queue
@@ -279,6 +280,19 @@ class _Watcher:
        # that survives defensive parsing is recorded here, including ones a
        # different host published. This is the external-observer record that
        # backstops the publisher's own "published" line if it never wrote one.
+        # Which dispatch this record vouches for. `dispatch_verified` says the
+        # token check above ran (the record had a dispatch_token and the
+        # event's data.dispatch_token matched it) after HMAC verification; on
+        # its own it names no dispatch, and this file is appended to across a
+        # job's attempts, so a reader waiting for a later dispatch would take
+        # an earlier attempt's record as its acknowledgement (review round 4).
+        # `dispatch_fingerprint` is SHA-256 of the token that was verified -
+        # the same identity the local path compares in events.log, minus the
+        # token itself, which the audit redactor masks by key name and which a
+        # 128-bit uuid4 nonce is not recoverable from. `attempt_id` is kept for
+        # readers; it is not the identity, since a token can be re-minted
+        # within an attempt. A reader that demands a token must match the
+        # fingerprint; a record without one proves nothing about identity.
        mqtt_common.append_event(jid, {
            "event": "received",
            "source_event": payload.get("event"),
@@ -286,6 +300,11 @@ class _Watcher:
            "topic": msg.topic,
            "timestamp": payload.get("timestamp"),
            "detail": payload.get("detail", ""),
+            "dispatch_verified": expected_disp_tok is not None,
+            "dispatch_fingerprint": (hashlib.sha256(expected_disp_tok.encode("utf-8")).hexdigest()
+                                     if expected_disp_tok else None),
+            "attempt_id": (data.get("attempt_id") if isinstance(data, dict) and data.get("attempt_id") is not None
+                           else self.attempts.get(jid)),
        })
        self.events.put((msg.topic, payload))

@@ -47,15 +47,42 @@ BACKOFF_MAX_S = 16.0
 CALLER_SCOPE_VARS = ("HERDR_WORKSPACE_ID", "HERDR_PANE_ID", "HERDR_TAB_ID")


+def caller_server(environ=None):
+    """The server whose pane this process runs in, "" when not in a pane.
+
+    HERDR_SESSION when present; otherwise the shape of HERDR_SOCKET_PATH, which
+    herdr exports to every pane (HERDR_SESSION is absent in a default-server
+    pane on herdr 0.8.2 - the orchestrator's usual home, and where the #23 leak
+    comes from). Same convention as lib.sh's _herdr_caller_server:
+
+        .../.config/herdr/herdr.sock                 -> "default"
+        .../.config/herdr/sessions/<name>/herdr.sock -> <name>
+
+    Matched by shape, not by $HOME. Anything else is "" (not in a pane).
+    """
+    src = os.environ if environ is None else environ
+    sess = (src.get("HERDR_SESSION") or "").strip()
+    if sess:
+        return sess
+    sock = (src.get("HERDR_SOCKET_PATH") or "").strip()
+    parts = sock.split("/")
+    if len(parts) >= 4 and parts[-1] == "herdr.sock":
+        if parts[-3:-1] == [".config", "herdr"]:
+            return "default"
+        if len(parts) >= 5 and parts[-4:-2] == ["herdr", "sessions"] and parts[-5] == ".config":
+            return parts[-2]
+    return ""
+
+
 def scoped_env(session=None, environ=None):
    """The environment for a wait aimed at `session`.

    Unchanged unless this process runs inside a pane of another server: only
    then are the caller's own workspace/pane/tab ids meaningless for the call
-    and dropped. No pane (no HERDR_SESSION) and same-server keep every value.
+    and dropped. No pane and same-server keep every value.
    """
    src = os.environ if environ is None else environ
-    caller = (src.get("HERDR_SESSION") or "").strip()
+    caller = caller_server(src)
    target = (session or "").strip()
    # `default` is the same sentinel build_argv reads: no --session flag, so
    # herdr answers on the ambient server - the caller's own.
@@ -60,7 +60,8 @@ If any check fails → abort with a non-zero exit and report the reason (automat
  - examples: `landing-page-refer-landing-page-creator-claude`, `paper-pdf2md-creator-agy`
  - never re-derive this by hand — source lib.sh and call the function
 - **wrapper script** (claude only): `~/.local/bin/<workspace-slug>-creator-claude`
-  - contents: herdr new-session with `claude` inside, auto-handles trust/bypass dialogs
+  - contents: herdr new-session with `claude` inside; trust/bypass dialogs are
+    handled by the skill for every agent (see `--no-auto-trust` below)
  - see `<workdir>/agent_sessions.md` for the canonical wrapper template

 ## Herdr Session Isolation (격리 세션)
@@ -97,6 +98,38 @@ The bootstrapped server receives a **sanitized copy of the caller's environment*
   bash scripts/create_session.sh --workspace /path/to/project --agent claude --role developer --onboard
   ```

+   **Delivery is judged by the agent's own `started` event** (#23), within the
+   job's `ack_timeout_sec` (default 180 s) — not by what the pane looked like two
+   seconds after the keys were sent. The script no longer publishes `started` on
+   the agent's behalf, so a job that shows `started` really was picked up. A
+   send_keys rc 4 ("Enter not seen accepted", where a still-starting TUI lands) is
+   a warning; rc 1/2/3 still abort at once. No `started` within the ack timeout
+   marks the job `error` and warns, but **keeps the session**: the pane is up and
+   the row is valid, only the handshake never landed, and killing a live agent
+   over a slow onboarding would destroy more than it protects (the same rule
+   this script applies to a TUI that never signalled ready). Re-send with
+   `multi-agent-mux-delegate-job submit` once the agent is up; the prompt is
+   never re-injected automatically. `MAM_ONBOARD_ACK_S` overrides how long that
+   wait is — distinct from the wrapper's `MAM_INJECT_ACK_GRACE_S`.
+
+   **Start-up dialogs** are answered for every agent, not only claude (#24). The
+   blocking-modal signatures and their accept keys live in each agent adapter
+   (`lib_py/agents/adapters/*.py`, published as `MAM_STARTUP_DIALOGS`), and they
+   are answered while waiting for the TUI rather than after it. `--no-auto-trust`
+   leaves directory-trust dialogs for a human to accept and still answers other
+   blocking notices (for example codex's update prompt, where Skip changes
+   nothing). Trust matters most for **codex**: trusting a directory makes it load
+   that project's local config, hooks and exec policy, so `--no-auto-trust` is the
+   switch to reach for when the workspace is not yours. Re-sending is rate-limited
+   by `MAM_DIALOG_RESEND_S` (default 5 s) because a `--source visible` snapshot
+   keeps showing a dialog for a while after it was accepted.
+
+   **Exit codes**: 42 = the TUI never became ready and the process is gone;
+   43 = the rollback could not kill the herdr session and it is still running.
+   On 43 the message names the session and the manual cleanup command; the YAML
+   row keeps `status: running` with `last_visible_status: rollback-failed` and no
+   `delegate_job_id`, and the onboarding job is closed as `cancelled`.
+
 Why use `--herdr-session`?

 - By default, all skills target `default` herdr session socket — fine for single-workspace use.
@@ -17,6 +17,8 @@
 #   2  = invalid args
 #   3  = herdr session already exists (use multi-agent-mux-resume or delete first)
 #   4  = agent-sessions.yaml append failure
+#   42 = agent TUI never became ready and the process is gone
+#   43 = rollback could not kill the herdr session; it is still running (#23)
 set -euo pipefail

 _script_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
@@ -329,8 +331,12 @@ cleanup_herdr_on_error() {
  # Release the onboarding job: `cancelled` is a terminal transition the
  # registry knows, and it says what happened - the job was never really
  # dispatched. Best effort; a failure here must not stop the rollback.
-  if [ -n "${DELEGATE_JOB_ID:-}" ]; then
+  # `completed`/`error`/`cancelled` are terminal and published exactly once, so
+  # a job this run already closed is left alone - a second one is refused by
+  # the registry anyway ("conflicting cancelled after terminal error").
+  if [ -n "${DELEGATE_JOB_ID:-}" ] && [ "${JOB_TERMINALIZED:-0}" != "1" ]; then
    delegate_publish_event "$DELEGATE_JOB_ID" cancelled "rolled back by create_session" "${DELEGATE_JOB_TOKEN:-}" "${DELEGATE_JOB_ATTEMPT:-}" || true
+    JOB_TERMINALIZED=1
  fi

  if [ "$alive" = "1" ]; then
@@ -608,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.
@@ -643,6 +639,7 @@ Task: $SUBMIT_JOB_PROMPT"
  # text never appeared - none of which a later `started` can repair.
  if [ "$rc" -ne 0 ] && [ "$rc" -ne 4 ]; then
    delegate_publish_event "$DELEGATE_JOB_ID" error "instruction injection failed (prompt-lock, rc=$rc)" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT"
+    JOB_TERMINALIZED=1
    exit 1
  fi
  if [ "$rc" -eq 4 ]; then
@@ -652,10 +649,38 @@ Task: $SUBMIT_JOB_PROMPT"
  # ISSUE-23: this script no longer publishes `started` on the agent's behalf.
  # It used to, which made the registry show a started job even when the agent
  # never saw the prompt, and left any started-based verdict meaningless.
-  if ! delegate_wait_started "$DELEGATE_JOB_ID" "${DELEGATE_JOB_ACK_S:-180}" "$DELEGATE_JOB_TOKEN"; then
-    delegate_publish_event "$DELEGATE_JOB_ID" error "onboarding delivery unconfirmed: no started within ${DELEGATE_JOB_ACK_S:-180}s" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT"
-    echo "ERROR: '$SESSION_NAME' did not acknowledge the onboarding prompt within ${DELEGATE_JOB_ACK_S:-180}s" >&2
-    exit 1
+  #
+  # How long to wait for the handshake: MAM_ONBOARD_ACK_S, else the wrapper's
+  # MAM_INJECT_ACK_GRACE_S, else the job's own ack timeout. The create-specific
+  # variable comes first so this wait can be shortened - in a sandbox with no
+  # real agent, for instance - without also retuning the delegate-job
+  # wrapper's injection ack, which is a different wait with its own tests.
+  ONBOARD_ACK_S="${MAM_ONBOARD_ACK_S:-${MAM_INJECT_ACK_GRACE_S:-${DELEGATE_JOB_ACK_S:-180}}}"
+  if ! delegate_wait_started "$DELEGATE_JOB_ID" "$ONBOARD_ACK_S" "$DELEGATE_JOB_TOKEN"; then
+    # The job failed, the session did not: the pane is up and the row is valid,
+    # only the handshake never landed. Killing a live agent over a slow or
+    # silent onboarding would destroy more than it protects, and it would
+    # contradict this script's own rule for the same situation one step
+    # earlier - a TUI that never signalled ready is preserved for diagnosis,
+    # 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.
+    # 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
  fi
  # Trigger immediate priority reconcile cycle asynchronously
  # Issue #21 phase 2 (A8): the control root's reconcile/watchdog, never the
@@ -633,41 +633,33 @@ _disk_status() {
 # ISSUE-18: delivery is confirmed by the protocol, not the screen. After the
 # prompt was handed over (herdr accepted the submission, or the dispatch was
 # published for a remote session) the job's own `started` must arrive within
-# the ack timeout T: the registry leaves `pending` (the agent's publish_event
-# syncs it on this host), a started line lands in events.log, or the
-# subscriber prints a started line (a remote agent - no shared disk). No
-# `started` within T is a delivery failure: fail closed, never re-inject
-# (single-injection contract). The status is not pinned to this dispatch and
-# may come from an external writer; the terminal handling after this still
-# keeps such an error/cancelled as it is rather than turning it into success.
+# the ack timeout T. The evidence is lib.sh's delegate_started_recorded - the
+# same predicate create_session's onboarding ack uses, so the two waits share
+# one contract (ack-finding req. 2, review R1): the local publisher's line in
+# events.log, or the subscriber's HMAC-verified, dispatch-attributed audit
+# record of a `started` it received (a remote agent has no shared disk). The
+# registry status is not evidence - a record leaves `pending` when another
+# writer closes it (error, cancelled) or another dispatch claims it, none of
+# which says this agent saw this prompt. No `started` within T is a delivery
+# failure: fail closed, never re-inject (single-injection contract). A job
+# that is already terminal is handled by the terminal logic after this, not
+# turned into a success here.
 _wait_started() {
-  local job_id="$1" ack_s="${2:-$ACK_TIMEOUT}" sub_log="${3:-}" deadline wall st
+  local job_id="$1" ack_s="${2:-$ACK_TIMEOUT}" sub_log="${3:-}" deadline wall
+  # sub_log is retained for callers' signatures; the subscriber's evidence is
+  # read from its audit record, which is attributable, not from its console.
  deadline=$(( $(date +%s) + ack_s ))
  # 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"'"'
  local budget=$(( deadline - $(date +%s) ))
  while :; do
-    st="$(_disk_status "$job_id")"
-    if [ -n "$st" ] && [ "$st" != "pending" ]; then
-      echo "inject: job $job_id advanced to '$st' in the registry; treating the prompt as delivered and continuing." >&2
-      return 0
-    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
+    # publish_event records `started` locally before its wire publish and
+    # before the status sync, so a slow broker cannot delay the acknowledgement.
+    if MAM_JOBS_DIR="$REGISTRY_DIR" delegate_started_recorded "$job_id" "${_disp_tok:-}" "${LOGS_DIR_ABS:-}"; then
      echo "inject: job $job_id recorded 'started' in events.log; treating the prompt as delivered and continuing." >&2
      return 0
    fi
-    if [ -n "$sub_log" ] && [ -f "$sub_log" ] && grep -Eq '^[^ ]+  job=[^ ]+  seq=[^ ]+  started( |$)' "$sub_log" 2>/dev/null; then
-      echo "inject: subscriber received started for job $job_id; treating the prompt as delivered and continuing." >&2
-      return 0
-    fi
    if [ "$(date +%s)" -ge "$deadline" ]; then
      ACK_WAITED_S="$budget"
      return 1
@@ -101,17 +101,11 @@ case "$AGENT" in
 esac

 # 4. Spawn new herdr session + run agent with the saved id
-case "$AGENT" in
-  claude)
-    START_CMD="herdr new-session -d -s \"$SESSION_NAME\" -x 140 -y 40 -c \"$WORKSPACE\" \"$CMD_FULL\""
-    eval "$START_CMD"
-    # auto-handle trust / bypass dialogs
-    handle_startup_dialogs "$SESSION_NAME" 20
-    ;;
-  agy|hermes|grok|opencode|codex)
 eval "herdr new-session -d -s \"$SESSION_NAME\" -x 140 -y 40 -c \"$WORKSPACE\" \"$CMD_FULL\""
-    ;;
-esac
+# Issue #24: start-up dialogs (trust prompts, blocking notices) are answered for
+# every agent from the adapter's signatures - the same call create makes.
+# --no-auto-trust leaves a directory-trust dialog to a human (MAM_AUTO_TRUST=0).
+handle_startup_dialogs "$SESSION_NAME" 20 "$AGENT"

 # 4. Update agent-sessions.yaml: status running, last_visible_status
 # (Also automatically publishes a `progress --detail "resumed"` event to the multi-agent-mux-delegate-job registry if a delegate_job_id exists)
@@ -13,6 +13,8 @@ Usage: $0 --workspace <path> --agent <claude|agy|hermes|grok|opencode|codex> --s

 Options:
  --herdr-session NAME specify isolated herdr session name (alias: --herdr-server)
+  --no-auto-trust      do not answer a start-up directory-trust dialog (default: answer it;
+                       other blocking notices are always answered) - same switch as create
  --dry-run            Simulates resume flow (resolves binary, environment) without writing
                       any updates to YAML or DB. Safe to execute inside active write transactions.
 EOF
@@ -25,6 +27,10 @@ HERDR_SERVER_OPT=""
 HERDR_WORKSPACE_OPT=""

 DRY_RUN=0
+# Issue #24: a resumed CLI shows the same directory-trust dialog as a fresh
+# one, so resume takes the same switch as create (MAM_AUTO_TRUST is what
+# handle_startup_dialogs reads).
+AUTO_TRUST=1

 while [ $# -gt 0 ]; do
  case "$1" in
@@ -33,6 +39,8 @@ while [ $# -gt 0 ]; do
    --session)   SESSION_NAME="$2"; shift 2 ;;
    --herdr-session|--herdr-server) HERDR_SERVER_OPT="$2"; shift 2 ;;
    --herdr-workspace) HERDR_WORKSPACE_OPT="$2"; shift 2 ;;
+    --auto-trust) AUTO_TRUST=1; shift ;;
+    --no-auto-trust) AUTO_TRUST=0; shift ;;
    --dry-run)   DRY_RUN=1; shift ;;
    -h|--help)   usage; exit 0 ;;
    *) echo "ERROR: unknown arg: $1" >&2; exit 2 ;;
@@ -46,6 +54,7 @@ case "$AGENT" in
  *) echo "ERROR: unsupported agent: $AGENT" >&2; exit 2 ;;
 esac
 [ -n "$SESSION_NAME" ] || { echo "ERROR: --session required" >&2; exit 2; }
+export MAM_AUTO_TRUST="$AUTO_TRUST"

 # Issue #21 phase 2 (A11): a session resumes only in the directory it ran in
 # (its conversation is keyed by that cwd; a class A agent would otherwise start
@@ -226,10 +235,10 @@ fi

 # 4. Spawn new agent session (delegates to herdr translation shim)
 _herdr new-session -d -s "$SESSION_NAME" -c "$WORKSPACE" "$CP_PREFIX$CMD_FULL"
-if [ "$AGENT" = "claude" ]; then
-  # auto-handle trust / bypass dialogs
-  handle_startup_dialogs "$SESSION_NAME" 20
-fi
+# Issue #24: every agent, the same call create makes. The signatures and accept
+# keys come from the agent adapter; this used to run only for claude, so a
+# resumed agy/grok/codex sat on its trust prompt.
+handle_startup_dialogs "$SESSION_NAME" 20 "$AGENT"

 # Wait for TUI readiness or let it settle
 sleep 2
@@ -227,6 +227,12 @@ def mam_sandbox(tmp_path, monkeypatch, request):
    # (`reconcile.sh --subscribe --idle-timeout 0`) from a sandbox. Tests that
    # exercise the daemon itself override this with "0" or delenv.
    monkeypatch.setenv("MAM_NO_SUBSCRIBER_DAEMON", "1")
+    # Issue #23: onboarding delivery now waits for the agent's own `started`.
+    # No sandbox has a real agent to publish one, so without a short grace
+    # every create_session test would sit out the full 180 s ack timeout.
+    # Create-specific on purpose: MAM_INJECT_ACK_GRACE_S is the delegate-job
+    # wrapper's knob and its own tests assert on the value they pass.
+    monkeypatch.setenv("MAM_ONBOARD_ACK_S", "3")
    monkeypatch.delenv("HERDR_SESSION_NAME", raising=False)
    monkeypatch.delenv("HERDR_SERVER_NAME", raising=False)
    monkeypatch.delenv("HERDR_WORKSPACE_ID", raising=False)
@@ -261,13 +261,21 @@ def test_wrapper_honours_registry_ack_over_unconfirmed_inject(tmp_path, mock_her
        time.sleep(0.1)
    assert jid, "job was not registered"

-    # The agent acknowledges the job out of band, as codex does via publish_event.
-    ack = subprocess.run(
-        [sys.executable, str(REGISTRY_PY), "--registry-dir", str(reg_dir),
-         "update", "--job", jid, "--status", "running"],
-        capture_output=True, text=True, cwd=str(tmp_path), env=env,
+    # The agent acknowledges the job out of band, as codex does via
+    # publish_event - the real producer, with the dispatch token the wrapper
+    # handed it. (Issue #23 review R1: a bare registry status flip is not an
+    # acknowledgement any more; only the agent's own `started` naming this
+    # dispatch is. publish_event writes that line locally before its wire
+    # publish, so an unreachable broker does not hide it.)
+    tok = json.loads((reg_dir / f"{jid}.json").read_text()).get("dispatch_token") or ""
+    subprocess.run(
+        [sys.executable, str(PUBLISH_PY), "--registry-dir", str(reg_dir), "--job", jid,
+         "--event", "started", "--expect-dispatch-token", tok, "--attempt-id", "1"],
+        capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60,
    )
-    assert ack.returncode == 0, ack.stderr
+    started_lines = [l for l in (reg_dir / f"{jid}.events.log").read_text().splitlines()
+                     if '"started"' in l]
+    assert started_lines and tok in started_lines[-1], started_lines

    stdout, stderr = proc.communicate(timeout=90)
    assert "treating the prompt as delivered and continuing" in stderr, f"ack guard did not fire:\n{stderr}\n{stdout}"
@@ -1225,11 +1225,11 @@ def test_acceptance5_remote_runner_completes_a_job_over_the_broker_only(tmp_path
                pytest.skip("broker unreachable when publishing the dispatch (no CONNACK); the runner never saw the job")
            _skip_if_broker_did_not_deliver(reg, runner_log)
        assert proc.returncode == 0, f"rc={proc.returncode}\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
-        # delivery is confirmed by the protocol: the subscriber saw `started`, or
-        # the registry left pending - to running, or straight to completed when
-        # the broker delivered only the retained terminal event
-        assert ("subscriber received started" in stderr or "advanced to 'running'" in stderr
-                or "advanced to 'completed'" in stderr), stderr
+        # delivery is confirmed by the protocol: the subscriber's attributed
+        # audit record of the remote `started` (review R1: the registry status
+        # is no longer evidence - it can leave pending for reasons that say
+        # nothing about this dispatch)
+        assert "recorded 'started'" in stderr, stderr
        assert _job(reg, jid)["status"] == "completed", \
            f"the subscriber mirrors the remote terminal into the registry\n{stdout}\n{stderr}\nrunner:\n{runner_log.read_text()}"
        assert _job(reg, jid)["transport"] == "remote"
@@ -17,6 +17,7 @@ agy/grok/codex/opencode TUI; no such TUI can be started here.
 import json
 import os
 import subprocess
+import sys
 import textwrap

 import pytest
@@ -34,7 +35,7 @@ def _run_lib(tmp_path, body, env=None):
    script = 'source "%s"\n_init_herdr_isolation\n%s\n' % (lib_path, body)
    run_env = dict(os.environ)
    for var in ("HERDR_SESSION", "HERDR_SESSION_NAME", "HERDR_WORKSPACE_ID",
-                "HERDR_PANE_ID", "HERDR_TAB_ID"):
+                "HERDR_PANE_ID", "HERDR_TAB_ID", "HERDR_SOCKET_PATH"):
        run_env.pop(var, None)
    run_env.update(env or {})
    return subprocess.run(["bash", "-c", script], capture_output=True,
@@ -109,6 +110,16 @@ CROSS = {"HERDR_SESSION": "default", "HERDR_SESSION_NAME": "mam-isolated",
         "HERDR_WORKSPACE_ID": "w5", "HERDR_PANE_ID": "w5:p2",
         "HERDR_TAB_ID": "w5:t1"}

+# The orchestrator pane as the live trial of this branch measured it (herdr
+# 0.8.2): no HERDR_SESSION at all, the default server's socket, and that
+# server's own ids. This is the env in which the HERDR_SESSION-based verdict
+# was always "not crossing" and #23 reproduced unchanged.
+DEFAULT_SOCK = "/home/someone/.config/herdr/herdr.sock"
+NAMED_SOCK = "/home/someone/.config/herdr/sessions/%s/herdr.sock"
+TRIAL_PANE = {"HERDR_SOCKET_PATH": DEFAULT_SOCK, "HERDR_WORKSPACE_ID": "w6",
+              "HERDR_PANE_ID": "w6:p1", "HERDR_TAB_ID": "w6:t1",
+              "HERDR_SESSION_NAME": "multi-agent-mux"}
+

 def test_s1_cross_server_call_drops_the_caller_panes_scope_ids(
        mam_sandbox, mock_herdr, mock_agents, monkeypatch):
@@ -189,12 +200,89 @@ def test_s2b_a_caller_outside_any_pane_keeps_the_explicit_scope(
    assert res.returncode == 0, res.stdout + res.stderr


+def test_s1c_the_real_default_server_pane_is_a_crossing(
+        mam_sandbox, mock_herdr, mock_agents, monkeypatch):
+    """S1c (trial finding, req. 4): the caller's server comes from the socket
+    path when HERDR_SESSION is absent. In the measured orchestrator pane the
+    three ids must be dropped and the agent on the named server found."""
+    tmp_path = mam_sandbox
+    env_log = _install_env_recorder(tmp_path, monkeypatch)
+    _seed_agent(mock_herdr, tmp_path, "trial-agy", "w1", "w1:p5")
+
+    res = _run_lib(tmp_path, 'herdr has-session -t trial-agy', env=TRIAL_PANE)
+    assert res.returncode == 0, "the trial's exact failure: " + res.stdout + res.stderr
+
+    calls = _recorded(env_log)
+    assert calls, "the recorder saw no herdr call"
+    for call in calls:
+        assert call["HERDR_WORKSPACE_ID"] is None, call
+        assert call["HERDR_PANE_ID"] is None, call
+        assert call["HERDR_TAB_ID"] is None, call
+
+
+def test_s1d_the_verdict_itself_on_the_trial_pane(mam_sandbox):
+    """S1d: the function the trial printed, in the trial's env: crossing."""
+    res = _run_lib(mam_sandbox,
+                   'echo "caller=$(_herdr_caller_server)"\n'
+                   'if _herdr_call_is_cross_server; then echo VERDICT=crossing; '
+                   'else echo VERDICT=not-crossing; fi',
+                   env=TRIAL_PANE)
+    assert "caller=default" in res.stdout, res.stdout + res.stderr
+    assert "VERDICT=crossing" in res.stdout, res.stdout + res.stderr
+
+
+def test_s1e_a_named_server_pane_calling_its_own_server_keeps_its_scope(
+        mam_sandbox, mock_herdr, mock_agents):
+    """S1e (req. 4): a caller in a named server's pane (socket under
+    sessions/<name>/) targeting that same server is not a crossing - the
+    exported workspace id still scopes, as in S2."""
+    tmp_path = mam_sandbox
+    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
+    same = {"HERDR_SOCKET_PATH": NAMED_SOCK % "mam-isolated",
+            "HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"}
+    res = _run_lib(tmp_path, 'echo "caller=$(_herdr_caller_server)"\nherdr has-session -t creator-agy-01',
+                   env=same)
+    assert "caller=mam-isolated" in res.stdout, res.stdout + res.stderr
+    assert res.returncode == 1, "same server: the explicit w2 scope must still hide a w1 agent"
+
+
+def test_s1f_a_named_server_pane_calling_another_named_server_is_a_crossing(
+        mam_sandbox, mock_herdr, mock_agents):
+    """S1f: named socket, different named target, no HERDR_SESSION - crossing."""
+    tmp_path = mam_sandbox
+    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
+    other = {"HERDR_SOCKET_PATH": NAMED_SOCK % "mam-other",
+             "HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"}
+    res = _run_lib(tmp_path, 'herdr has-session -t creator-agy-01', env=other)
+    assert res.returncode == 0, res.stdout + res.stderr
+
+
+def test_s1g_herdr_session_wins_over_the_socket_when_both_are_present(mam_sandbox):
+    """S1g (req. 1): HERDR_SESSION, when a herdr exports it, is authoritative
+    over the socket-derived name."""
+    res = _run_lib(mam_sandbox, 'echo "caller=$(_herdr_caller_server)"',
+                   env={"HERDR_SESSION": "from-herdr", "HERDR_SOCKET_PATH": DEFAULT_SOCK})
+    assert "caller=from-herdr" in res.stdout, res.stdout + res.stderr
+
+
+def test_s1h_an_unrecognised_socket_layout_is_not_in_a_pane(mam_sandbox, mock_herdr, mock_agents):
+    """S1h: a socket path that is neither layout yields no caller server, so
+    an exported scope is respected (today's behaviour), never guessed away."""
+    tmp_path = mam_sandbox
+    _seed_agent(mock_herdr, tmp_path, "creator-agy-01", "w1", "w1:p5")
+    res = _run_lib(tmp_path, 'echo "caller=[$(_herdr_caller_server)]"\nherdr has-session -t creator-agy-01',
+                   env={"HERDR_SOCKET_PATH": "/tmp/odd/herdr.sock",
+                        "HERDR_SESSION_NAME": "mam-isolated", "HERDR_WORKSPACE_ID": "w2"})
+    assert "caller=[]" in res.stdout, res.stdout + res.stderr
+    assert res.returncode == 1, res.stdout + res.stderr
+
+
 def test_s3_shim_contract_regressions_still_pass(mam_sandbox):
    """S3: H-18 / H-21 / H-23 are the tests that fix today's scoping meaning.
    Run them explicitly as this change's regression guard."""
    repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    res = subprocess.run(
-        [os.path.join(repo, ".venv", "bin", "python"), "-m", "pytest", "-q",
+        [sys.executable, "-m", "pytest", "-q",
         "-p", "no:cacheprovider",
         "tests/test_herdr_shim_contract.py", "-k",
         "h18 or h21 or h22 or h23"],
@@ -270,7 +358,6 @@ def test_s12_mam_herdr_never_aborts_its_caller_under_set_e(
 def test_s13_the_loop_watcher_applies_the_same_rule(mam_sandbox):
    """S13 (pure unit): watcher.py builds its own herdr argv and never goes
    through mam_herdr, so it needs the same rule."""
-    import sys
    repo = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    sys.path.insert(0, os.path.join(repo, ".agents", "skills"))
    from lib_py.loop import watcher
@@ -291,6 +378,18 @@ def test_s13_the_loop_watcher_applies_the_same_rule(mam_sandbox):
    none_target = watcher.scoped_env(None, environ=caller_pane)
    assert none_target["HERDR_WORKSPACE_ID"] == "w5"

+    # trial finding: no HERDR_SESSION, the caller's server comes from the socket
+    trial = {"HERDR_SOCKET_PATH": DEFAULT_SOCK, "HERDR_WORKSPACE_ID": "w6",
+             "HERDR_PANE_ID": "w6:p1", "HERDR_TAB_ID": "w6:t1"}
+    assert watcher.caller_server(trial) == "default"
+    assert "HERDR_WORKSPACE_ID" not in watcher.scoped_env("multi-agent-mux", environ=trial)
+    named_same = {"HERDR_SOCKET_PATH": NAMED_SOCK % "iso-x", "HERDR_WORKSPACE_ID": "w2"}
+    assert watcher.caller_server(named_same) == "iso-x"
+    assert watcher.scoped_env("iso-x", environ=named_same)["HERDR_WORKSPACE_ID"] == "w2"
+    assert "HERDR_WORKSPACE_ID" not in watcher.scoped_env("iso-y", environ=named_same)
+    assert watcher.caller_server({"HERDR_SOCKET_PATH": "/tmp/odd/herdr.sock"}) == ""
+    assert watcher.caller_server({"HERDR_SESSION": "s", "HERDR_SOCKET_PATH": DEFAULT_SOCK}) == "s"
+
    seen = {}

    def _run(argv, **kw):
@@ -331,7 +430,11 @@ echo "rc=$?"
 ''' % (lib, pane_file, keys_log, prelude, timeout, agent)
    run_env = dict(os.environ)
    run_env.pop("MAM_AUTO_TRUST", None)
-    run_env.update(env or {})
+    for k, v in (env or {}).items():
+        if v == "":
+            run_env.pop(k, None)     # "" means: not in the environment at all
+        else:
+            run_env[k] = v
    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
                         cwd=str(tmp_path), env=run_env)
    sent = keys_log.read_text().splitlines() if keys_log.exists() else []
@@ -372,7 +475,7 @@ def test_s4b_every_adapter_publishes_the_fact(mam_sandbox):
    seen = {}
    for agent in ("claude", "agy", "grok", "codex", "opencode", "hermes"):
        out = subprocess.run(
-            [os.path.join(repo, ".venv", "bin", "python"), "-m", "lib_py.agents",
+            [sys.executable, "-m", "lib_py.agents",
             "facts", agent],
            capture_output=True, text=True, cwd=repo,
            env=dict(os.environ, PYTHONPATH=skills)).stdout
@@ -417,6 +520,47 @@ def test_s5c_readiness_is_judged_by_the_ready_tokens_not_by_the_dialog_leaving(
    assert "rc=0" in res.stdout, res.stdout + res.stderr


+READY_FRAMES = [
+    ("claude", "Welcome to Claude Code\n❯ "),
+    ("agy", "Antigravity Ready in /tmp\n> "),
+    ("grok", "Grok Ready\n>>> "),
+    ("codex", "Codex\nAsk Codex to do anything"),
+    ("opencode", "OpenCode\nAsk anything"),
+]
+
+
+@pytest.mark.parametrize("agent,frame", READY_FRAMES)
+def test_s5d_a_ready_first_frame_returns_within_one_iteration_with_no_env(
+        mam_sandbox, agent, frame):
+    """S5d (review B-1): called on its own - as resume_session.sh calls it,
+    with no wait_for_tui_ready before it and no MAM_*READY* in the environment
+    - the handler must recognise the adapter's ready banner on the first frame
+    and return at once, not idle through its timeout. `sleep` is stubbed so
+    the iteration count is observable and the run is fast; base returned in
+    0.03s, the version under review took the full timeout."""
+    res, sent = _run_dialogs(
+        mam_sandbox, agent, frame, timeout=20,
+        env={"MAM_STRONG_READY_TOKENS": "", "MAM_READY_TOKENS": "", "MAM_WEAK_READY_TOKENS": ""},
+        prelude='_iters=0\nsleep() { _iters=$((_iters + 1)); }\n'
+                'trap \'echo "iterations=$_iters"\' EXIT')
+    assert "rc=0" in res.stdout, res.stdout + res.stderr
+    assert sent == [], sent
+    iters = int(res.stdout.split("iterations=")[1].split()[0])
+    assert iters <= 1, "the handler idled %d iterations on a ready screen" % iters
+
+
+def test_s5e_caller_set_ready_tokens_are_not_overwritten(mam_sandbox):
+    """S5e: the gap-fill must not clobber tokens a caller set on purpose -
+    wait_for_tui_ready's own evaluation, or a test steering readiness."""
+    res, sent = _run_dialogs(
+        mam_sandbox, "grok", "Grok Ready\n>>> ", timeout=2,
+        env={"MAM_STRONG_READY_TOKENS": "NonExistentStrong", "MAM_READY_TOKENS": "NonExistentStrong"},
+        prelude='_iters=0\nsleep() { _iters=$((_iters + 1)); }\n'
+                'trap \'echo "iterations=$_iters"\' EXIT')
+    iters = int(res.stdout.split("iterations=")[1].split()[0])
+    assert iters >= 2, "the caller's tokens were replaced by the adapter's: " + res.stdout
+
+
 def test_s6_no_auto_trust_leaves_trust_dialogs_alone_but_answers_notices(
        mam_sandbox):
    """S6 (stubbed): --no-auto-trust. codex trust is left to a human with a
@@ -471,27 +615,73 @@ 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()
+    # 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:
-        ev.write_text(json.dumps(
-            {"job": job_id, "event": "started", "dispatch_token": started_token}) + "\n")
+            cmd += ["--expect-dispatch-token", started_token, "--attempt-id", "1"]
+        pub = subprocess.run(cmd, capture_output=True, text=True, env=env, timeout=60)
+        # a mismatched token is refused by the producer's own CAS check; that
+        # is fine here - the point is what (if anything) reached the log
+        if started_token in (None, "tok-1"):
+            assert pub.returncode == 0, pub.stdout + pub.stderr
+    if status != "pending":
+        rec = jobs / ("%s.json" % job_id)
+        d = json.loads(rec.read_text())
+        d["status"] = status
+        rec.write_text(json.dumps(d))
    return jobs


-def _run_delivery(tmp_path, inject_rc, job_id="job1", **seed):
+def _run_delivery(tmp_path, inject_rc, job_id="job1", seed_events=True, **seed):
    """Run create_session.sh's own delivery block with a stubbed injection."""
    lib = tmp_path / ".agents" / "skills" / "lib.sh"
    block = _extract(_create_sh(tmp_path), "# Inject instructions into the herdr pane",
                     "delegate_wait_started", close="fi")
+    if seed_events:
        _seed_job(tmp_path, job_id, **seed)
    pub_log = tmp_path / "published.txt"
    script = '''
@@ -532,26 +722,437 @@ def test_s7b_delivery_succeeds_on_the_agents_own_started(mam_sandbox):
    assert published == [], published


-def test_s7c_delivery_succeeds_when_the_registry_left_pending(mam_sandbox):
-    """S7c: publish_event syncs the record on this host, so a status past
-    `pending` is the same evidence."""
-    res, _ = _run_delivery(mam_sandbox, 0, status="running")
+def test_s7c_a_started_that_was_followed_by_a_terminal_status_still_counts(mam_sandbox):
+    """S7c: a job that started and then finished (or failed) fast still shows
+    its started line - the evidence is the line, whatever the status became."""
+    for status in ("running", "completed", "error"):
+        res, _ = _run_delivery(mam_sandbox, 0, status=status, started_token="tok-1")
+        assert "DELIVERED" in res.stdout, (status, res.stdout + res.stderr)
+
+
+@pytest.mark.parametrize("status", ["error", "cancelled"])
+def test_s7f_a_job_closed_by_someone_else_is_not_an_acknowledgement(mam_sandbox, status):
+    """S7f (review R1 + req. 4): a record can leave `pending` because another
+    writer closed it. That is not evidence - the wait runs to its deadline
+    rather than returning early (the 'no started event naming this dispatch'
+    path, never the DELIVERED-without-warning path of S7c) - and, the job
+    being already terminal, nothing is published over it (req. 4)."""
+    res, published = _run_delivery(mam_sandbox, 0, status=status)
+    assert published == [], (status, published, res.stderr)
+    assert "no started event naming this dispatch" in res.stderr, res.stderr
+    assert "already '%s'" % status in res.stderr, res.stderr
+
+
+def test_s7g_another_dispatch_running_is_not_this_dispatchs_started(mam_sandbox):
+    """S7g (review R1): `running` set by a different dispatch's claim says
+    nothing about this prompt."""
+    res, published = _run_delivery(mam_sandbox, 0, status="running")
+    assert published == ["error"], (published, res.stderr)
+
+
+def test_s7h_the_line_is_parsed_not_pattern_matched(mam_sandbox):
+    """S7h (review N-4), the one SYNTHETIC line in this file, on purpose: the
+    parser must not depend on key order. Every other started line here comes
+    from the real producer (S7b/S7c/S7e/S8, S7i-S7k)."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1")
+    reordered = {"data": {"attempt_id": 1, "dispatch_token": "tok-1"}, "event": "started",
+                 "job_id": "job1"}
+    (jobs / "job1.events.log").write_text(json.dumps(reordered) + "\n")
+    res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
    assert "DELIVERED" in res.stdout, res.stdout + res.stderr
+    assert published == [], published


-def test_s7d_no_started_within_t_is_a_delivery_failure(mam_sandbox):
-    """S7d: fail closed, and say so on the job."""
+# ---- trial-2 regression: the real agent must be able to satisfy the ack ----
+
+def test_s7i_the_old_create_session_command_shape_could_never_be_acknowledged(mam_sandbox):
+    """S7i (trial 2): the command create_session used to hand the agent had no
+    --expect-dispatch-token, so the REAL producer wrote a started with no
+    token, and a token-checking waiter never accepted it: 180 s idle, then a
+    false 'unconfirmed' over a finished onboarding. Pinned as the negative it
+    is, so the fixture can never drift back to faking the token."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1", publish_args=[])          # old shape: no token args
+    line = json.loads((jobs / "job1.events.log").read_text().splitlines()[0])
+    assert line["event"] == "started" and "dispatch_token" not in line.get("data", {}), line
+    res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
+    assert published == ["error"], (published, res.stderr)
+
+
+def test_s7j_the_shared_builders_command_is_acknowledged_by_the_real_producer(mam_sandbox):
+    """S7j (trial 2, the fix): run exactly the command delegate_publish_cmd
+    hands the agent, through the real publish_event.py, and the waiter is
+    satisfied on the first probe."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1")
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    env = _producer_env(tmp_path)
+    env["MAM_JOBS_DIR"] = str(jobs)
+    env["WORKSPACE_ROOT"] = str(tmp_path)
+    script = ('source "%s"\n'
+              'cmd="$(delegate_publish_cmd job1 tok-1 1)"\n'
+              'echo "CMD=$cmd"\n'
+              'eval "$cmd --event started" >/dev/null\n'
+              'delegate_started_recorded job1 tok-1 && echo RECORDED\n') % lib
+    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                         cwd=str(tmp_path), env=env, timeout=120)
+    assert "RECORDED" in res.stdout, res.stdout + res.stderr
+    cmd = res.stdout.split("CMD=")[1].splitlines()[0]
+    assert "--expect-dispatch-token tok-1" in cmd and "--attempt-id 1" in cmd, cmd
+    line = json.loads((jobs / "job1.events.log").read_text().splitlines()[-1])
+    assert line["data"]["dispatch_token"] == "tok-1", line
+
+
+def test_s7k_create_session_hands_the_agent_the_builders_command(mam_sandbox):
+    """S7k: the onboarding instruction is built by delegate_publish_cmd with
+    the job's token and attempt id - not by a hand-rolled copy that can drift."""
+    src = open(_create_sh(mam_sandbox)).read()
+    assert 'pub="$(delegate_publish_cmd "$DELEGATE_JOB_ID" "$DELEGATE_JOB_TOKEN" "$DELEGATE_JOB_ATTEMPT")"' in src
+    assert 'printf \'%q \' "$py_bin" "$pub_script" --registry-dir' not in src
+
+
+def test_s7l_the_builder_matches_the_wrappers_pub_cmd_argument_for_argument(mam_sandbox):
+    """S7l (req. 2, one contract): the wrapper's _pub_cmd and lib.sh's
+    delegate_publish_cmd produce the same argv for the same job."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1")
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+    env = _producer_env(tmp_path)
+    env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
+                "MAM_DELEGATE_JOB_NO_MAIN": "1"})
+    script = ('source "%s"\n'
+              'source "%s"\n'
+              'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; JOB_ID=job1; LOGS_DIR_ABS="$DELEGATE_JOB_LOGS_DIR"\n'
+              'echo "W=$(_pub_cmd --expect-dispatch-token tok-1 --attempt-id 1)"\n'
+              'echo "L=$(delegate_publish_cmd job1 tok-1 1)"\n') % (lib, wrapper, jobs)
+    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                         cwd=str(tmp_path), env=env, timeout=120)
+    w = res.stdout.split("W=")[1].splitlines()[0].split()
+    l = res.stdout.split("L=")[1].splitlines()[0].split()
+    # same script, registry, job and token/attempt flags; the interpreter path
+    # may differ in spelling (the wrapper resolves it relative), so compare from
+    # the script name on
+    def tail(argv):
+        i = next(i for i, a in enumerate(argv) if a.endswith("publish_event.py"))
+        return argv[i:]
+    assert tail(w) == tail(l), (w, l)
+
+
+def test_s7m_a_job_the_agent_already_finished_is_not_marked_error(mam_sandbox):
+    """S7m (req. 4): with no started naming this dispatch but the job already
+    `completed`, create_session leaves it alone - no `error` over a terminal
+    state (the trial-2 'refused error ... after terminal completed')."""
+    res, published = _run_delivery(mam_sandbox, 0, status="completed")
+    assert published == [], (published, res.stderr)
+    assert "already 'completed'" in res.stderr, res.stderr
+    assert "DELIVERED" in res.stdout
+
+
+def _wrapper_wait(tmp_path, jobs, disp_tok, ack_s=2, logs_dir=None):
+    """Run the wrapper's _wait_started exactly as the wrapper runs it."""
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+    env = _producer_env(tmp_path)
+    env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
+                "MAM_DELEGATE_JOB_NO_MAIN": "1"})
+    script = ('source "%s"\nsource "%s"\nset +e\n'
+              'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; ACK_TIMEOUT=%d; START_TIME=$(date +%%s); TIMEOUT=60\n'
+              'LOGS_DIR_ABS="%s"\n'
+              '_disp_tok=%s; _wait_started job1 %d; echo "rc=$?"\n'
+              ) % (lib, wrapper, jobs, ack_s, logs_dir or env["DELEGATE_JOB_LOGS_DIR"], disp_tok, ack_s)
+    return subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                          cwd=str(tmp_path), env=env, timeout=120)
+
+
+@pytest.mark.parametrize("terminal", ["error", "cancelled"])
+def test_w1_wrapper_a_job_closed_without_a_started_is_not_acknowledged(mam_sandbox, terminal):
+    """W1 (review R1, real producer, record NOT reset): the agent never
+    published started; someone published `%s`. The record left `pending`.
+    The wrapper's wait must still say no - as create's does."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1", publish_args=None)
+    env = _producer_env(tmp_path)
+    subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
+                    "--event", terminal, "--detail", "closed by another writer",
+                    "--expect-dispatch-token", "tok-1", "--attempt-id", "1"],
+                   check=True, capture_output=True, text=True, env=env, timeout=60)
+    assert json.loads((jobs / "job1.json").read_text())["status"] == terminal
+    res = _wrapper_wait(tmp_path, jobs, "tok-1")
+    assert "rc=1" in res.stdout, res.stdout + res.stderr
+    assert "treating the prompt as delivered" not in res.stderr, res.stderr
+
+
+def test_w2_wrapper_another_dispatchs_started_and_running_is_not_this_ones(mam_sandbox):
+    """W2 (review R1, real producer, record NOT reset): started published
+    under token A (record -> running), then the job was re-dispatched with
+    token B. Waiting for B must not accept A's started, nor the `running`."""
+    tmp_path = mam_sandbox
+    jobs = tmp_path / ".mam" / "jobs"; jobs.mkdir(parents=True, exist_ok=True)
+    env = _producer_env(tmp_path)
+    subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
+                    "--prompt", "x", "--agent", "claude-code", "--agent-session", "herdr:x",
+                    "--dispatch-token", "tok-A", "--job-id", "job1", "--ack-timeout", "2"],
+                   check=True, capture_output=True, text=True, env=env)
+    subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
+                    "--event", "started", "--expect-dispatch-token", "tok-A", "--attempt-id", "1"],
+                   check=True, capture_output=True, text=True, env=env, timeout=60)
+    rec = jobs / "job1.json"; d = json.loads(rec.read_text())
+    assert d["status"] == "running"
+    d["dispatch_token"] = "tok-B"; rec.write_text(json.dumps(d))     # the re-dispatch; status untouched
+    res = _wrapper_wait(tmp_path, jobs, "tok-B")
+    assert "rc=1" in res.stdout, res.stdout + res.stderr
+
+
+def test_w3_wrapper_this_dispatchs_started_and_running_is_acknowledged(mam_sandbox):
+    """W3 (real producer, record NOT reset): the positive twin of W2."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1", started_token="tok-1")
+    assert json.loads((jobs / "job1.json").read_text())["status"] == "running"
+    res = _wrapper_wait(tmp_path, jobs, "tok-1")
+    assert "rc=0" in res.stdout, res.stdout + res.stderr
+
+
+def test_w4_both_waits_agree_on_every_case(mam_sandbox):
+    """W4 (ack-finding req. 2): create's delegate_wait_started and the
+    wrapper's _wait_started answer the same on the four reviewer cases."""
+    tmp_path = mam_sandbox
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    cases = []
+    for label, status, started_tok, wait_tok in (
+            ("error-no-started", "error", None, "tok-1"),
+            ("cancelled-no-started", "cancelled", None, "tok-1"),
+            ("other-dispatch-started", "running", "tok-1", "tok-B"),
+            ("own-started", "running", "tok-1", "tok-1")):
+        jobs = _seed_job(tmp_path, "job1", status=status, started_token=started_tok)
+        env = _producer_env(tmp_path)
+        env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path)})
+        c = subprocess.run(["bash", "-c", 'source "%s"; delegate_wait_started job1 1 %s; echo "rc=$?"' % (lib, wait_tok)],
+                           capture_output=True, text=True, cwd=str(tmp_path), env=env, timeout=60)
+        w = _wrapper_wait(tmp_path, jobs, wait_tok, ack_s=1)
+        cases.append((label, c.stdout.strip().splitlines()[-1], w.stdout.strip().splitlines()[-1]))
+    for label, c_rc, w_rc in cases:
+        assert c_rc == w_rc, (label, cases)
+    assert [c for _, c, _ in cases] == ["rc=1", "rc=1", "rc=1", "rc=0"], cases
+
+
+def test_w5_a_remote_agents_started_counts_only_through_the_subscribers_verified_record(mam_sandbox):
+    """W5 (remote route): a remote agent leaves no line in events.log; the
+    subscriber writes an audit record after HMAC verification and after
+    dropping any dispatch_token mismatch, and marks it dispatch_verified.
+    SYNTHETIC on purpose: the record is written in exactly the shape
+    mqtt_common.append_event produces (a live subscriber needs a broker).
+    With a token demanded, only a verified record counts."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1")
+    logs = tmp_path / ".mam" / "delegate_job_logs"
+    (logs / "job1").mkdir(parents=True, exist_ok=True)
+    audit = logs / "job1" / "events.ndjson"
+
+    def rec(**kw):
+        base = {"event": "received", "source_event": "started", "seq": 1, "topic": "t",
+                "timestamp": "2026-09-22T00:00:00Z", "detail": "", "logged_at": "x"}
+        base.update(kw)
+        return json.dumps(base) + "\n"
+
+    import hashlib
+    fp = lambda t: hashlib.sha256(t.encode()).hexdigest()
+    audit.write_text(rec(dispatch_verified=False, dispatch_fingerprint=None))   # untokened: not attributed
+    assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+    audit.write_text(rec(source_event="progress", dispatch_verified=True, dispatch_fingerprint=fp("tok-1")))
+    assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout   # not a started
+    audit.write_text(rec(dispatch_verified=True))                                # verified, no identity: fail closed
+    assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+    audit.write_text(rec(dispatch_verified=True, dispatch_fingerprint=fp("tok-OLD")))   # another dispatch's
+    assert "rc=1" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+    audit.write_text(rec(dispatch_verified=True, dispatch_fingerprint=fp("tok-1")))
+    assert "rc=0" in _wrapper_wait(tmp_path, jobs, "tok-1", ack_s=1, logs_dir=str(logs)).stdout
+
+
+def test_w6_the_subscriber_stamps_dispatch_verified_on_what_it_records(mam_sandbox):
+    """W6: the real subscriber code path - a _Watcher with an expected
+    dispatch token records a matching started as dispatch_verified, and drops
+    a mismatching one before recording anything."""
+    import types
+    sys.path.insert(0, SKILLS)
+    from lib_py.delegation import job_subscriber, mqtt_common
+    logs = mam_sandbox / ".mam" / "delegate_job_logs"
+    w = job_subscriber._Watcher({"job1"}, {"job1": None}, {"job1": 0}, {"job1": "tok-1"}, {"job1": 1})
+    msg = types.SimpleNamespace(topic="python/mqtt/jobs/job1/events")
+    good = {"schema_version": 1, "seq": 1, "job_id": "job1", "event": "started", "timestamp": "t",
+            "detail": "", "data": {"dispatch_token": "tok-1", "attempt_id": 1}}
+    bad = dict(good, seq=2, data={"dispatch_token": "tok-OLD", "attempt_id": 1})
+    for payload in (good, bad):
+        w._on_event(msg, payload, "job1") if hasattr(w, "_on_event") else w.on_message(None, None, types.SimpleNamespace(topic=msg.topic, payload=json.dumps(payload).encode()))
+    recs = list(mqtt_common.iter_logged_events("job1", str(logs)))
+    started = [r for r in recs if r.get("source_event") == "started"]
+    assert len(started) == 1 and started[0]["dispatch_verified"] is True, recs
+    import hashlib
+    assert started[0]["dispatch_fingerprint"] == hashlib.sha256(b"tok-1").hexdigest(), started[0]
+    assert started[0]["attempt_id"] == 1, started[0]
+    assert "tok-1" not in json.dumps(started[0]), "the token itself must never reach the audit log"
+
+
+# --------------------------------------------------------------------------
+# review round 4 (planner-reviewer-claude-01 B-1, planner-reviewer-codex-01):
+# the remote audit record must name the dispatch it vouches for. Real signed
+# producer, real _Watcher.on_message, real audit writer, real re-dispatch
+# (registry operator write minting a new token) - no record is edited by hand.
+# --------------------------------------------------------------------------
+
+def _remote_fixture(tmp_path, monkeypatch):
+    import test_issue18_liveness_contract as live
+    jobs = tmp_path / "jobs"
+    env = live._env(tmp_path, dict(live.UNREACHABLE, MAM_DELEGATE_JOB_NO_MAIN="1",
+                                   MAM_JOBS_DIR=str(jobs)))
+    for key in ("MAM_ENV_FILE", "DELEGATE_JOB_LOGS_DIR", "WORKSPACE_ROOT"):
+        monkeypatch.setenv(key, env[key])
+    sys.path.insert(0, str(live.SKILLS))
+    from lib_py.delegation import job_subscriber, mqtt_common, publish_event, registry
+    return live, env, jobs, job_subscriber, mqtt_common, publish_event, registry
+
+
+def _receive(publish_event, watcher, jid, token, attempt, seq, secret):
+    """A signed `started` from the real producer, handed to the real subscriber."""
+    import types
+    payload = publish_event.build_payload(jid, seq, "started", "",
+                                          {"dispatch_token": token, "attempt_id": attempt}, secret)
+    watcher.on_message(None, None, types.SimpleNamespace(
+        topic="python/mqtt/jobs/%s/events" % jid, payload=json.dumps(payload).encode()))
+
+
+def _both_waits(live, env, tmp_path, jobs, jid, token):
+    """create's delegate_wait_started and the wrapper's _wait_started, same inputs."""
+    import shlex
+    script = (
+        'source %s\nset +e\nPY=%s\nREGISTRY_DIR=%s\nACK_TIMEOUT=0\nTIMEOUT=60\nSTART_TIME=$(date +%%s)\n'
+        'LOGS_DIR_ABS=%s\n_disp_tok=%s\n'
+        'delegate_wait_started %s 0 "$_disp_tok"; echo CREATE_RC=$?\n'
+        '_wait_started %s 0; echo WRAPPER_RC=$?\n'
+    ) % (shlex.quote(str(live.WRAPPER)), shlex.quote(sys.executable), shlex.quote(str(jobs)),
+         shlex.quote(env["DELEGATE_JOB_LOGS_DIR"]), shlex.quote(token), jid, jid)
+    res = subprocess.run(["bash", "-c", script], env=env, cwd=str(tmp_path),
+                         capture_output=True, text=True, timeout=60)
+    out = res.stdout + res.stderr
+    c = int(out.split("CREATE_RC=")[1].split()[0]); w = int(out.split("WRAPPER_RC=")[1].split()[0])
+    return c, w, out
+
+
+@pytest.mark.parametrize("scenario,expected", [
+    ("current", 0), ("old_audit", 1), ("old_audit_and_stale_wire", 1),
+    ("new_dispatch_received", 0), ("invalid_hmac", 1),
+])
+def test_w7_remote_audit_is_bound_to_the_dispatch_being_waited_for(tmp_path, monkeypatch, scenario, expected):
+    """W7 (codex's five cases, real paths throughout). Attempt 1 under token A
+    leaves a verified audit record; the operator re-dispatches (attempt 2,
+    new token B, nothing edited by hand); waiting for B must not be answered
+    by A's record, even when B's subscriber correctly dropped A's replayed
+    packet - and B's own verified started must still be accepted. Both waits
+    must agree."""
+    live, env, jobs, job_subscriber, mqtt_common, publish_event, registry = _remote_fixture(tmp_path, monkeypatch)
+    jid, secret = "auditjob", "test-only-hmac-secret"
+    registry.register_job("remote audit", job_id=jid, registry_dir=str(jobs), auth_token=secret,
+                          dispatch_token="dispatch-A", transport="remote",
+                          broker={"host": "127.0.0.1", "port": 9, "tls": False})
+    w1 = job_subscriber._Watcher({jid}, {jid: secret}, {jid: 0}, {jid: "dispatch-A"}, {jid: 1})
+    _receive(publish_event, w1, jid, "dispatch-A", 1, 1, "wrong-secret" if scenario == "invalid_hmac" else secret)
+    recs = list(mqtt_common.iter_logged_events(jid, env["DELEGATE_JOB_LOGS_DIR"]))
+    assert sum(r.get("source_event") == "started" for r in recs) == (0 if scenario == "invalid_hmac" else 1), recs
+    token = "dispatch-A"
+    if scenario not in ("current", "invalid_hmac"):
+        job_subscriber._sync_remote_status(jid, "completed", str(jobs), token)
+        assert registry._operator_write(jid, str(jobs), status="pending", attempt_id=2, iteration=2) == 0
+        record = mqtt_common.load_job(jid, str(jobs))
+        token = record["dispatch_token"]
+        assert token != "dispatch-A" and record["attempt_id"] == 2, record
+        w2 = job_subscriber._Watcher({jid}, {jid: secret}, {jid: 0}, {jid: token}, {jid: 2})
+        if scenario == "old_audit_and_stale_wire":
+            _receive(publish_event, w2, jid, "dispatch-A", 1, 2, secret)     # A's packet replayed to B's subscriber
+            assert w2.events.empty(), "the subscriber must drop the stale dispatch"
+        elif scenario == "new_dispatch_received":
+            _receive(publish_event, w2, jid, token, 2, 2, secret)
+            assert not w2.events.empty()
+    assert not (jobs / ("%s.events.log" % jid)).exists(), "remote evidence comes only from the subscriber's audit"
+    c, w, out = _both_waits(live, env, tmp_path, jobs, jid, token)
+    assert (c, w) == (expected, expected), (scenario, c, w, out)
+
+
+def test_w8_a_record_without_identity_fails_closed_when_a_token_is_demanded(tmp_path, monkeypatch):
+    """W8: an audit record from before dispatch_fingerprint existed (verified,
+    no fingerprint) names no dispatch: rejected when a token is demanded,
+    accepted when none is (an untokened legacy job), exactly as before."""
+    live, env, jobs, job_subscriber, mqtt_common, publish_event, registry = _remote_fixture(tmp_path, monkeypatch)
+    jid = "legacyjob"
+    registry.register_job("legacy", job_id=jid, registry_dir=str(jobs), auth_token=None,
+                          dispatch_token="dispatch-L", transport="remote",
+                          broker={"host": "127.0.0.1", "port": 9, "tls": False})
+    mqtt_common.append_event(jid, {"event": "received", "source_event": "started", "seq": 1,
+                                   "topic": "t", "timestamp": "t", "detail": "",
+                                   "dispatch_verified": True}, env["DELEGATE_JOB_LOGS_DIR"])
+    c, w, out = _both_waits(live, env, tmp_path, jobs, jid, "dispatch-L")
+    assert (c, w) == (1, 1), ("token demanded: a record with no identity must not count", out)
+    c, w, out = _both_waits(live, env, tmp_path, jobs, jid, "")
+    assert (c, w) == (0, 0), ("no token demanded: compatibility path", out)
+
+
+def test_s7n_the_wrappers_wait_started_accepts_the_real_producers_line(mam_sandbox):
+    """S7n (req. 2): the delegate-job wrapper's _wait_started, on the same
+    real-producer started (token in data), accepts it; on a stale token it
+    does not - via the shared predicate."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1", started_token="tok-1")
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    wrapper = tmp_path / ".agents" / "skills" / "multi-agent-mux-delegate-job" / "multi-agent-mux-delegate-job"
+    env = _producer_env(tmp_path)
+    env.update({"MAM_JOBS_DIR": str(jobs), "WORKSPACE_ROOT": str(tmp_path),
+                "MAM_DELEGATE_JOB_NO_MAIN": "1"})
+    # the record stays exactly as the real producer left it (`running`):
+    # review R1 found the previous version resetting it to `pending`, which
+    # hid the wrapper's status bypass behind the predicate
+    script = ('source "%s"\nsource "%s"\n'
+              'PY="$(_delegate_py_bin)"; REGISTRY_DIR="%s"; ACK_TIMEOUT=2; START_TIME=$(date +%%s); TIMEOUT=60\n'
+              '_disp_tok=tok-1;   _wait_started job1 2 && echo ACCEPT_OWN\n'
+              '_disp_tok=tok-X;   _wait_started job1 2 || echo REJECT_STALE\n') % (lib, wrapper, jobs)
+    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                         cwd=str(tmp_path), env=env, timeout=120)
+    assert "ACCEPT_OWN" in res.stdout, res.stdout + res.stderr
+    assert "REJECT_STALE" in res.stdout, res.stdout + res.stderr
+
+
+def test_s7d_no_started_marks_the_job_but_keeps_the_session(mam_sandbox):
+    """S7d: the job failed, the session did not. The handshake is marked
+    `error` and warned about loudly, but a live agent is not killed over a
+    slow onboarding - the same rule this script already applies one step
+    earlier to a TUI that never signalled ready."""
    res, published = _run_delivery(mam_sandbox, 0)
-    assert res.returncode == 1, res.stdout + res.stderr
-    assert "DELIVERED" not in res.stdout
+    assert res.returncode == 0, res.stdout + res.stderr
    assert published == ["error"], published
+    assert "did not acknowledge the onboarding prompt" in res.stderr
+    assert "preserved" in res.stderr


 def test_s7e_a_started_from_another_dispatch_is_not_evidence(mam_sandbox):
    """S7e: a loop job keeps one events.log across attempts, so only a started
-    carrying THIS dispatch token counts."""
-    res, _ = _run_delivery(mam_sandbox, 0, started_token="tok-OTHER")
-    assert res.returncode == 1, res.stdout + res.stderr
+    carrying THIS dispatch token counts. The stale line is made by the real
+    producer under the record's *previous* token, then the record is
+    re-dispatched (token rotated) before the waiter looks."""
+    tmp_path = mam_sandbox
+    jobs = tmp_path / ".mam" / "jobs"
+    jobs.mkdir(parents=True, exist_ok=True)
+    env = _producer_env(tmp_path)
+    subprocess.run([sys.executable, REGISTRY_PY, "--registry-dir", str(jobs), "register",
+                    "--prompt", "x", "--agent", "claude-code", "--agent-session", "herdr:x",
+                    "--dispatch-token", "tok-OLD", "--job-id", "job1", "--ack-timeout", "2"],
+                   check=True, capture_output=True, text=True, env=env)
+    subprocess.run([sys.executable, PUBLISH_PY, "--registry-dir", str(jobs), "--job", "job1",
+                    "--event", "started", "--expect-dispatch-token", "tok-OLD", "--attempt-id", "1"],
+                   check=True, capture_output=True, text=True, env=env, timeout=60)
+    rec = jobs / "job1.json"
+    d = json.loads(rec.read_text()); d["dispatch_token"] = "tok-1"     # the re-dispatch; status untouched
+    rec.write_text(json.dumps(d))
+    res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
+    assert published == ["error"], (published, res.stderr)


 def test_s8_rc4_is_a_warning_and_the_started_wait_decides(mam_sandbox):
@@ -564,8 +1165,8 @@ def test_s8_rc4_is_a_warning_and_the_started_wait_decides(mam_sandbox):
    assert published == [], published

    res, published = _run_delivery(mam_sandbox, 4)
-    assert res.returncode == 1, res.stdout + res.stderr
    assert published == ["error"], published
+    assert "did not acknowledge the onboarding prompt" in res.stderr


@pytest.mark.parametrize("rc", [1, 2, 3])
@@ -0,0 +1,127 @@
+"""Issue #24, resume path: start-up dialogs are answered for every agent.
+
+resume_session.sh called handle_startup_dialogs only for claude and passed no
+agent, so a resumed agy/grok/codex sat on its directory-trust prompt exactly as
+the create path did before this branch. These fix the call contract resume
+now shares with create.
+
+What is real: the shipped create/stop/resume scripts run end to end against
+the mock herdr in a sandbox. What is not: the mock's pane always shows a
+ready banner, so no test here drives a real trust dialog - the create-side
+tests (test_issue23_24_startup_path.py, S4-S6) fix which keys each dialog
+gets, and R4 shows the same shipped function is now reached from resume.
+"""
+
+import json
+import os
+import subprocess
+import time
+
+import pytest
+
+SCRIPTS = ("multi-agent-mux-create/scripts/create_session.sh",
+           "multi-agent-mux-stop/scripts/stop_session.sh",
+           "multi-agent-mux-resume/scripts/resume_session.sh")
+
+
+def _script(tmp_path, rel):
+    return str(tmp_path / ".agents" / "skills" / rel)
+
+
+def _create_and_stop(tmp_path, agent, name):
+    ws = str(tmp_path / ("ws_" + name))
+    os.makedirs(ws, exist_ok=True)
+    create, stop, _ = (_script(tmp_path, s) for s in SCRIPTS)
+    subprocess.run(["bash", create, "--workspace", ws, "--agent", agent,
+                    "--role", "creator", "--session", name, "--no-onboard"],
+                   check=True, cwd=str(tmp_path), capture_output=True, text=True)
+    subprocess.run(["bash", stop, "--session", name, "--agent", agent],
+                   check=True, cwd=str(tmp_path), capture_output=True, text=True)
+    return ws
+
+
+def _resume(tmp_path, agent, name, ws, extra=()):
+    _, _, resume = (_script(tmp_path, s) for s in SCRIPTS)
+    return subprocess.run(["bash", resume, "--workspace", ws, "--agent", agent,
+                           "--session", name, *extra],
+                          cwd=str(tmp_path), capture_output=True, text=True)
+
+
+def _record_dialog_calls(tmp_path):
+    """Override handle_startup_dialogs in the SANDBOX copy of lib.sh (never the
+    repo's) so the call resume makes - session, timeout, agent, MAM_AUTO_TRUST
+    - is written down. Bash keeps the last definition, so appending wins."""
+    log = tmp_path / "dialog_calls.log"
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    with open(lib, "a") as f:
+        f.write('\nhandle_startup_dialogs() {\n'
+                '  printf \'%%s\\n\' "sess=$1 timeout=${2:-} agent=${3:-<none>} '
+                'auto_trust=${MAM_AUTO_TRUST:-<unset>}" >> "%s"\n'
+                '}\n' % log)
+    return log
+
+
+def _calls(log):
+    return log.read_text().splitlines() if log.exists() else []
+
+
+@pytest.mark.parametrize("agent", ["agy", "codex"])
+def test_r1_resume_answers_dialogs_for_a_non_claude_agent_and_names_it(
+        mam_sandbox, mock_herdr, mock_agents, agent):
+    """R1: the gate is gone and the agent is passed - the two halves of the
+    create contract resume was missing."""
+    tmp_path = mam_sandbox
+    name = "test-r1-" + agent
+    ws = _create_and_stop(tmp_path, agent, name)
+    log = _record_dialog_calls(tmp_path)   # armed after create, so only resume is recorded
+    res = _resume(tmp_path, agent, name, ws)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert _calls(log) == ["sess=%s timeout=20 agent=%s auto_trust=1" % (name, agent)]
+
+
+def test_r2_claude_keeps_its_dialog_handling(mam_sandbox, mock_herdr, mock_agents):
+    """R2: claude is still handled, now by the same call as everyone else."""
+    tmp_path = mam_sandbox
+    ws = _create_and_stop(tmp_path, "claude", "test-r2-claude")
+    log = _record_dialog_calls(tmp_path)
+    res = _resume(tmp_path, "claude", "test-r2-claude", ws)
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert _calls(log) == ["sess=test-r2-claude timeout=20 agent=claude auto_trust=1"]
+
+
+def test_r3_no_auto_trust_reaches_the_dialog_handler(mam_sandbox, mock_herdr, mock_agents):
+    """R3: --no-auto-trust on resume sets MAM_AUTO_TRUST=0 for the handler,
+    the same switch with the same meaning as create's."""
+    tmp_path = mam_sandbox
+    ws = _create_and_stop(tmp_path, "codex", "test-r3-codex")
+    log = _record_dialog_calls(tmp_path)
+    res = _resume(tmp_path, "codex", "test-r3-codex", ws, extra=("--no-auto-trust",))
+    assert res.returncode == 0, res.stdout + res.stderr
+    assert _calls(log) == ["sess=test-r3-codex timeout=20 agent=codex auto_trust=0"]
+
+
+def test_r4_the_shipped_handler_really_runs_for_a_non_claude_resume(
+        mam_sandbox, mock_herdr, mock_agents):
+    """R4 (no override): with the real handle_startup_dialogs, a codex resume
+    completes and the pane is read after the spawn - the handler was reached
+    and returned on the ready banner. Before this change no pane read happened
+    between new-session and the YAML update for a non-claude agent."""
+    tmp_path = mam_sandbox
+    ws = _create_and_stop(tmp_path, "codex", "test-r4-codex")
+    before = len(json.load(open(mock_herdr)).get("calls", []))
+    t0 = time.monotonic()
+    res = _resume(tmp_path, "codex", "test-r4-codex", ws)
+    elapsed = time.monotonic() - t0
+    assert res.returncode == 0, res.stdout + res.stderr
+    # Review B-1: the handler used to idle its whole 20 s timeout here because
+    # it had no ready tokens on the resume path. rc 0 alone cannot tell early
+    # return from timeout; the clock can.
+    assert elapsed < 12, "resume took %.1fs: the dialog handler idled to its timeout" % elapsed
+    calls = json.load(open(mock_herdr)).get("calls", [])[before:]
+
+    def _has(c, *verbs):   # the mock records argv with a leading `--session <name>`
+        return any(tuple(c[i:i + len(verbs)]) == verbs for i in range(len(c)))
+
+    spawn = next(i for i, c in enumerate(calls) if _has(c, "agent", "start"))
+    reads = [c for c in calls[spawn + 1:] if _has(c, "agent", "read") or _has(c, "pane", "read")]
+    assert reads, "no pane read after the spawn: handle_startup_dialogs never ran\n%s" % calls
@@ -692,22 +692,90 @@ def test_g9_loop_package_imports_only_delegation_among_siblings():
    assert not offenders, offenders


+# The G9 guard reads CODE, not prose. Its claim is that no frozen lib.sh
+# helper is *invoked* from lib_py/loop; a comment or docstring that names
+# lib.sh (say, to point at the shell twin of a function so the two stay in
+# sync) invokes nothing. Every way Python could actually reach a lib.sh helper
+# - a path or `source` inside a subprocess argv, an identifier - lives in code
+# text, which is still scanned in full, so the guard loses none of its reach.
+# History: it first fired on `hashlib.sha256` (a word boundary fixed that),
+# then on a docstring cross-reference in watcher.py; both were mentions.
+_G9_PATTERNS = {
+    "ensure_subscriber_daemon": r"ensure_subscriber_daemon",
+    "preexec_fn": r"preexec_fn",
+    "lib.sh": r"\blib\.sh\b",     # word boundary: `hashlib.sha256` is not lib.sh
+}
+
+
+def _code_only(src):
+    """`src` as code only: the docstring expression statements are removed from
+    the AST and the rest is unparsed, which also drops comments (they are not
+    in the AST). Only the docstring *node* goes, so a statement that shares
+    its line - a docstring followed by `; call()` - is kept, as are string literals in code
+    (a path or `source` in a subprocess argv is exactly what must be caught).
+    Review R2: blanking the docstring's whole line range hid same-line code."""
+    import ast
+
+    tree = ast.parse(src)
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
+            body = getattr(node, "body", None) or []
+            first = body[0] if body else None
+            if (isinstance(first, ast.Expr) and isinstance(first.value, ast.Constant)
+                    and isinstance(first.value.value, str)):
+                del body[0]
+                if not body:
+                    body.append(ast.Pass())
+    return ast.unparse(tree)
+
+
+def _g9_offenders(src):
+    import re
+    code = _code_only(src)
+    return [name for name, pat in _G9_PATTERNS.items() if re.search(pat, code)]
+
+
 def test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper():
    """The two phase 1 handoffs stay dormant: preexec_fn is unreachable because
    nothing here spawns, and no frozen lib.sh helper is invoked."""
-    import re
-
-    # `lib.sh` needs a word boundary: a bare substring search also matches
-    # hashlib.sha256, which is how this guard first fired on innocent code.
-    patterns = {
-        "ensure_subscriber_daemon": re.compile(r"ensure_subscriber_daemon"),
-        "preexec_fn": re.compile(r"preexec_fn"),
-        "lib.sh": re.compile(r"\blib\.sh\b"),
-    }
    for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
        src = path.read_text(encoding="utf-8")
-        for name, pattern in patterns.items():
-            assert not pattern.search(src), "{} references {}".format(path.name, name)
+        hits = _g9_offenders(src)
+        assert not hits, "{} references {}".format(path.name, ", ".join(hits))
+
+
+def test_g9b_the_guard_still_catches_an_invocation_and_ignores_prose():
+    """The narrowing must not cost the guard its purpose: an invocation-shaped
+    reference in code text still trips it; the same words in a docstring or a
+    comment do not."""
+    invoked = (
+        'import subprocess\n'
+        'def go():\n'
+        '    return subprocess.run(["bash", "-c", "source .agents/skills/lib.sh; ensure_subscriber_daemon x"])\n'
+    )
+    assert set(_g9_offenders(invoked)) == {"lib.sh", "ensure_subscriber_daemon"}
+    path_only = 'CMD = ["bash", "/x/.agents/skills/lib.sh"]\n'
+    assert _g9_offenders(path_only) == ["lib.sh"]
+    prose = (
+        '"""Module doc: same convention as lib.sh\'s _herdr_caller_server."""\n'
+        'import hashlib\n'
+        'def f():\n'
+        '    """See lib.sh; never call ensure_subscriber_daemon from here."""\n'
+        '    # preexec_fn is documented in lib.sh\n'
+        '    return hashlib.sha256(b"x").hexdigest()\n'
+    )
+    assert _g9_offenders(prose) == []
+    # Review R2: code sharing a line with a docstring must still be seen -
+    # module, one-line body, one-line def, and the last line of a multi-line
+    # docstring. Only the docstring node is removed, never its line.
+    same_line = [
+        '"""module doc"""; ensure_subscriber_daemon()\n',
+        'def f():\n    """function doc"""; ensure_subscriber_daemon()\n',
+        'def f(): """function doc"""; ensure_subscriber_daemon()\n',
+        'def f():\n    """multi\n    line"""; ensure_subscriber_daemon()\n',
+    ]
+    for snippet in same_line:
+        assert _g9_offenders(snippet) == ["ensure_subscriber_daemon"], snippet


 # ==============================================================================
Author	SHA1	Message	Date
GodopuandClaude Fable 5.1	f4ba205c9f	fix(issue-23): the remote audit record names the dispatch it vouches for; a record without identity fails closed Review round 4 (planner-reviewer-claude-01 B-1, planner-reviewer-codex-01 P2, reached independently): dispatch_verified is a boolean. It says the subscriber's token check ran when the record was written, but names no dispatch, the token being masked by the audit redactor and attempt_id not written. events.ndjson is appended to across a job's attempts, so a started verified under attempt 1 satisfied every later attempt's wait - the very misattribution R1 forbids, alive on the remote path while the local path (data.dispatch_token in events.log) was already immune. codex reproduced it with the real signed producer, the real _Watcher, the real audit writer and the real operator re-dispatch; both waits answered 0 where 1 was due, even when attempt 2's subscriber had correctly dropped attempt 1's replayed packet. The subscriber now writes dispatch_fingerprint = SHA-256 of the token it verified, and attempt_id for readers. With a token demanded, the predicate accepts an audit record only if it is dispatch_verified AND its fingerprint equals SHA-256 of that token; a record with no fingerprint - written before the field existed - names nothing and is rejected on that path. Without a token, any received started counts, as before. Fingerprint over attempt_id as the identity, weighing the two proposals: the local path already compares the token, so hashing it keeps one identity axis for both sources; the token is a uuid4 nonce (registry.py:141), 128 bits, so its SHA-256 is not recoverable and it is a CAS fence rather than the HMAC secret in the first place; and attempt_id cannot tell apart a token re-minted within one attempt, where the token can. attempt_id is recorded because it is cheap and readable, not relied on. W7 pins codex's five cases end to end - current dispatch accepted, an old attempt's record rejected after a real re-dispatch, still rejected when the stale packet is replayed and dropped, the new dispatch's own started accepted, a bad HMAC never recorded - with create's wait and the wrapper's wait required to agree on each. W8 pins the fail-closed legacy record and the untokened compatibility path. W5 gains the no-fingerprint and other-dispatch cases; W6 checks the real _Watcher writes the fingerprint and never the token. S7e's status reset had no reason left and is gone. codex's own reproduction file, unmodified, passes 5/5 against this tree. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>	2026-09-22 17:52:51 +00:00
GodopuandClaude Fable 5.1	087eacaf21	fix(issue-23): both ack waits share one contract - the wrapper's status bypass is gone; the G9 guard removes only the docstring node Review round 3 (planner-reviewer-codex-01, reviewer-opencode-01): R1 blocking, R2 and the "ONE builder" wording non-blocking. All three reproduced; none rebutted. R1. _wait_started still returned 0 as soon as the record left `pending`, ahead of the shared predicate, so `error`, `cancelled` and another dispatch's `running` all acknowledged an injection that create's delegate_wait_started rejects - the two contracts the ack-finding brief said must not diverge. Both reviewers reproduced it with the real producer and no record manipulation; my S7n had hidden it by resetting the record to `pending`. The bypass and the console grep are gone. The wrapper's evidence is now delegate_started_recorded alone, the same call create makes. For a remote agent - no shared disk, so no line in events.log - the predicate reads the subscriber's audit record of the `started` it received. That record is attributable by construction: the subscriber verifies the HMAC and drops any event whose data.dispatch_token differs from the job's expected token before it writes anything. The token itself cannot go into the record (the audit redactor masks token-named keys), so the subscriber now stamps a boolean dispatch_verified, and with a token demanded only a verified record counts; an untokened legacy record fails closed. The registry status is never evidence; a terminal job is preserved by the terminal logic, not turned into an acknowledgement. Tests: W1/W2/W3 are the reviewers' three negatives and the positive, real producer, record left exactly as it produced it; W4 runs create's wait and the wrapper's wait side by side on all four and requires the same answer; W5 pins the remote route (a verified record counts, an unverified or non-started one does not); W6 drives the real _Watcher and shows it stamps the flag on a matching started and records nothing for a mismatch; S7n no longer resets the record. test_issue14's "registry ack over unconfirmed inject" produced its ack with a bare status flip - the very bypass - and now produces it the way an agent does, publish_event with the dispatch token; its scenario and every other assertion are unchanged. test_issue18's remote e2e accepted the status messages and now expects the shared one. R2. _code_only blanked a docstring's whole line range, so code sharing that line - `"""doc"""; call()` - vanished from the scan. It now removes only the docstring node from the AST and unparses the rest (comments drop out with it, code strings stay); test_g9b pins the four same-line shapes. The builder comment claimed one builder for create and the wrapper; the wrapper keeps _pub_cmd and S7l pins the two equal argument-for-argument. The comment now says that. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>	2026-09-22 16:43:00 +00:00
GodopuandClaude Fable 5.1	4805fd4511	test(issue-5): the G9 no-frozen-helper guard reads code, not comments or docstrings The full suite on `1667fda` showed exactly one failure beyond this host's baseline: test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper. Its `\blib\.sh\b` pattern matched a docstring in watcher.py that points at the shell twin of caller_server ("Same convention as lib.sh's _herdr_caller_server") - a mention written so the two implementations stay in sync, not a call. The guard's own docstring states its claim: no frozen lib.sh helper is invoked from lib_py/loop. Its one prior change (`803ba43`) added the word boundary after `hashlib.sha256` tripped it - the same class of false positive, handled by narrowing the matcher rather than by contorting innocent code. This continues that: the three patterns and the word boundary are unchanged, but they now run over the source with comments and docstrings blanked. Every way Python could actually reach a lib.sh helper - a path or `source` in a subprocess argv, an identifier - lives in code text, which is still scanned in full, so the guard keeps its reach. test_g9b pins both sides: an invocation-shaped string in code still trips it (lib.sh and ensure_subscriber_daemon both reported), a bare lib.sh path in a list literal still trips it, and the same words in a module docstring, a function docstring and a comment do not - with hashlib.sha256 present as the original regression. Rephrasing the docstring to dodge the literal was the other option and was not taken: it would hide a useful cross-reference to satisfy a matcher, which is the evasion shape this guard should not encourage, and it would leave the next honest comment to trip the same wire. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>	2026-09-22 15:32:40 +00:00
GodopuandClaude Fable 5.1	1667fda8dd	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>	2026-09-22 14:42:38 +00:00
GodopuandClaude Fable 5.1	f7cee00dec	fix(issue-23): the caller's server comes from HERDR_SOCKET_PATH, not only HERDR_SESSION The live trial of this branch found the cross-server verdict inert on this host: _herdr_call_is_cross_server read the caller's server from HERDR_SESSION, and herdr 0.8.2 does not export that variable to a pane of the default server - the orchestrator's usual home, and exactly the pane the #23 leak comes from. Every call was judged "not crossing", the three ids were never dropped, and all four trial onboardings ended "no started within 180s". My own pane sits in a named server and does carry HERDR_SESSION, which is how the assumption got past me. HERDR_SOCKET_PATH is exported to every pane and its layout names the server: .../.config/herdr/herdr.sock is the default server and .../.config/herdr/sessions/<name>/herdr.sock a named one - the convention `herdr session list --json` reports and the shim already uses to find a named server's socket. _herdr_caller_server prefers HERDR_SESSION when a herdr does export it, otherwise reads the socket's shape (suffix-matched, not $HOME- anchored), and yields "" for anything unrecognised so an exported scope is respected rather than guessed away. The rest of the rule is unchanged from `3bcaa39`: a `default` or empty target is the ambient server, no caller server means no pane, caller == target is the same server. watcher.caller_server applies the same rule for the loop's own herdr calls. Reproduced against the live server in the trial's exact env: `10a678c` answers "NOT crossing / has-session NOT FOUND"; this commit answers "caller server default / crossing / FOUND". S1c pins that env end to end with the recorder asserting the ids never reach herdr; S1d-S1h pin the verdict, the named-socket same-server and cross-server cases, HERDR_SESSION precedence, and the unrecognised layout. The `3bcaa39` tests are unchanged and still pass. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>	2026-09-22 14:13:38 +00:00
GodopuandClaude Fable 5.1	10a678c010	fix(issue-23-24): the dialog handler sees the ready banner on its own; only this dispatch's started is delivery evidence Review round 1 (planner-reviewer-claude-01, reviewer-agy-01, planner-reviewer-codex-01): one blocking defect and two P2s, all confirmed. B-1. handle_startup_dialogs took its ready tokens only from the environment, which wait_for_tui_ready fills when it evaluates the adapter facts. Called on its own - as resume_session.sh calls it, with no readiness wait before it - it had no tokens, its early-return branch could never fire, and every resume of every agent idled through the full 20 s timeout (base: 0.03 s). rc was 0 throughout, so nothing failed; the resume tests just took ~30 s each. _mam_load_dialog_facts now also takes MAM_STRONG_READY_TOKENS / MAM_READY_TOKENS from the facts, but only when the caller has not set them, so a caller-set value (wait_for_tui_ready's own, or a test steering readiness) is never clobbered - the concern that made it take a single fact in the first place. S5d pins one-iteration return on a ready first frame for five adapters with no environment; S5e pins that caller tokens survive; R4 gets a wall-clock cap so a timeout-idle can no longer pass as success. The old _MAM_READY_TOKENS_CLAUDE fallback was left defined but unread - removed. R1 / N-2. delegate_wait_started treated any status past `pending` as the agent's acknowledgement. codex reproduced three false positives: `error` and `cancelled` records with no started line, and a `running` record from another dispatch. The status branch is gone; the evidence is a started line carrying this dispatch's token, which the local publisher writes before it syncs the status, so nothing that really started is missed (S7c re-pinned on the line with running/completed/error statuses; S7f/S7g pin the three negatives). N-4, taken with it: the events.log line is parsed as JSON rather than pattern- matched. publish_event nests dispatch_token under `data`, and the old regex only matched because of key order (S7h, real line shape and a reordered one). R2 / N-1. Two tests hardcoded <repo>/.venv/bin/python and failed with FileNotFoundError in every review worktree. sys.executable. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>	2026-09-22 10:25:52 +00:00
GodopuandClaude Fable 5.1	0dfbb2cc8a	fix(issue-24): resume answers start-up dialogs for every agent, with the same switch as create resume_session.sh called handle_startup_dialogs only when AGENT=claude and passed no agent, so a resumed agy, grok or codex sat on its directory-trust prompt exactly as the create path did before this branch. The gate is gone and the agent is passed, which is the call create already makes. --no-auto-trust / MAM_AUTO_TRUST apply to resume too. A resumed CLI shows the same trust dialog as a fresh one, in the same directory, with the same consequence for codex (trusting loads the project's local config, hooks and exec policy), and handle_startup_dialogs already reads MAM_AUTO_TRUST - so the only question was whether resume exposes the switch or silently inherits the caller's environment. Exposing it keeps the two skills' contracts identical. The tests run the shipped create/stop/resume scripts against the mock herdr. The mock's pane always shows a ready banner, so they fix the call resume makes (session, timeout, agent, MAM_AUTO_TRUST) by overriding the handler in the sandbox copy of lib.sh, plus one unmodified run showing the real handler is reached for a non-claude agent; which keys a dialog gets is fixed on the create side (S4-S6). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>	2026-09-22 08:19:18 +00:00
GodopuandClaude Opus 5	ce802d21c8	docs(issue-23): the create skill documents the preserve-on-unacknowledged-onboarding rule and MAM_ONBOARD_ACK_S Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-09-22 06:24:33 +00:00
GodopuandClaude Opus 5	33b91fe0c8	fix(issue-23): an unacknowledged onboarding marks the job and keeps the session Two defects in the started-based verdict, both found by the full suite. The first was destructive. No `started` within T exited 1, which fired the rollback trap and killed the session. That is worse than the bug being fixed: create_session already has a rule for "the agent is not responding but the process is alive" one step earlier - a TUI that never signalled ready is preserved for diagnosis - and an onboarding that is merely slow is the same situation. The job is marked `error` and warned about loudly; the session, its row and the pane stay. MULTI_AGENT_RULES §6 has the GM gate on this job anyway, so an unacknowledged onboarding still blocks real work. It also broke 21 tests across 8 modules (create/sanity/tier2/tier3/tier4/ issue15/issue19/issue21) and added roughly an hour to the suite: no sandbox has an agent to publish `started`, so every create_session test sat out the full 180 s and then failed on the rollback. MAM_ONBOARD_ACK_S shortens that wait, and the sandbox fixture sets 3 s. It is deliberately not the wrapper's MAM_INJECT_ACK_GRACE_S - that is a different wait whose own tests assert on the value they pass, and reusing it made issue18's acceptance2 read "within 3s" instead of "within 4s". The second: the delivery failure published `error` and then the rollback published `cancelled` on the same job, which the registry refuses ("conflicting cancelled after terminal error"). Terminal events are published exactly once, so the trap now skips a job this run already closed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-09-22 06:24:06 +00:00
GodopuandClaude Opus 5	2cd20423ac	docs(issue-23-24): the create skill documents --no-auto-trust, the started-based onboarding verdict and exit 43 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>	2026-09-22 04:28:57 +00:00