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>

tests/test_issue23_24_startup_path.py

@@ -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
@@ -194,7 +195,7 @@ def test_s3_shim_contract_regressions_still_pass(mam_sandbox):
     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 +271,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
@@ -331,7 +331,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 +376,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 +421,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
@@ -487,12 +532,13 @@ def _seed_job(tmp_path, job_id, status="pending", started_token=None):
     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")
-    _seed_job(tmp_path, job_id, **seed)
+    if seed_events:
+        _seed_job(tmp_path, job_id, **seed)
     pub_log = tmp_path / "published.txt"
     script = '''
 set -euo pipefail
@@ -532,11 +578,46 @@ 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")
-    assert "DELIVERED" in res.stdout, res.stdout + res.stderr
+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): a record can leave `pending` because another writer
+    closed it. No started line from this agent means no delivery."""
+    res, published = _run_delivery(mam_sandbox, 0, status=status)
+    assert published == ["error"], (status, published, res.stderr)
+    assert "did not acknowledge the onboarding prompt" in res.stderr
+
+
+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_real_events_log_shape_is_read(mam_sandbox):
+    """S7h (review N-4): publish_event nests dispatch_token under `data` and
+    the line is JSON, so it is parsed, not pattern-matched. A line with the
+    keys in another order, and a token in either place, both count."""
+    tmp_path = mam_sandbox
+    jobs = _seed_job(tmp_path, "job1")
+    real = {"schema_version": 1, "seq": 1, "job_id": "job1", "event": "started",
+            "timestamp": "2026-09-22T08:04:26Z", "detail": "",
+            "data": {"dispatch_token": "tok-1", "attempt_id": 1}}
+    reordered = {"data": {"attempt_id": 1, "dispatch_token": "tok-1"}, "event": "started",
+                 "job_id": "job1"}
+    for line in (real, reordered):
+        (jobs / "job1.events.log").write_text(json.dumps(line) + "\n")
+        res, published = _run_delivery(mam_sandbox, 0, seed_events=False)
+        assert "DELIVERED" in res.stdout, (line, res.stdout + res.stderr)
+        assert published == [], published
 
 
 def test_s7d_no_started_marks_the_job_but_keeps_the_session(mam_sandbox):