fix(issue-23): onboarding delivery is judged by the agent's own started event, and a rollback verifies the kill

Plan commits 3 and 4 landed together: both rewrite the same region of
create_session.sh and share one test file, and splitting them afterwards would
have meant hunk-level surgery on already-verified code.

Delivery verdict (item 3). create_session --onboard called a healthy agent a
failure three seconds after registering the job: inject_instructions ->
send_keys_safe returns rc 4 when `agent prompt --wait --until working` fails
and the pane is still not working two seconds later, which is exactly where a
TUI that is still starting up lands. It never got the ISSUE-18 confirmation the
delegate-job wrapper has. rc 4 is a warning now and the agent's own `started`
decides, within the job's own ack timeout; rc 1/2/3 stay hard failures, since
they say the pane never quiesced, a dialog was blocking, or the text never
appeared, and no later `started` repairs those.

The self-published `started` had to go first. create_session published it on
the agent's behalf right after injection, so the registry showed a started job
whether or not the agent ever saw the prompt - with that line in place, no
started-based verdict can mean anything.

delegate_wait_started is lib.sh's counterpart to the wrapper's _wait_started
(plan item D10-b). The wrapper's own `submit` was the first choice, but it
blocks until the job reaches a terminal state, and create_session must return
as soon as delivery is confirmed; the on-disk evidence it reads - registry
status past `pending`, or a started line carrying this dispatch token - is
written by publish_event before its wire publish, so no broker is involved.

Rollback integrity (item 4). The trap ran `_herdr kill-session -t
"$SESSION_NAME" 2>/dev/null || true`, so a kill that closed nothing still
printed "rolling back" while the session kept running, and the row stayed
`running` pointing at the job the same rollback had just failed (#23 symptom
5). With item 1 in place this check is finally trustworthy: the kill used to
fail because the caller's workspace id hid the pane.

The trap now reads the kill's rc, confirms with has-session, closes the
onboarding job as `cancelled` (a terminal transition the registry knows, and
the truthful one - the job was never dispatched), and marks the row. A session
still alive exits 43 with the manual cleanup command, distinct from 42 (TUI
never ready).

The row keeps a schema-valid status rather than the plan's `error`, which
atomic_yaml.py rejects: `terminated` once the session is really gone, and
`running` when the kill failed, because it is still running. What makes that
case legible is the note, the cleared delegate_job_id and exit 43 - not a
status claiming the opposite of the truth. The row is never deleted; a deleted
row takes the evidence with it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

tests/test_issue23_24_startup_path.py

@@ -444,3 +444,224 @@ def test_s6b_create_session_no_longer_gates_dialogs_on_claude(mam_sandbox):
     assert call, "create_session no longer answers start-up dialogs at all"
     assert all('"$AGENT"' in l for l in call), call
     assert 'if [ "$AGENT" = "claude" ] && [ -z "$LAST_VISIBLE_STATUS_OVERRIDE" ]' not in src
+
+
+# --------------------------------------------------------------------------
+# item 3 — onboarding delivery is judged by the agent's own `started` (#23)
+# --------------------------------------------------------------------------
+
+def _create_sh(tmp_path):
+    return tmp_path / ".agents" / "skills" / "multi-agent-mux-create" / "scripts" / "create_session.sh"
+
+
+def _extract(path, start_marker, end_marker, close=None):
+    """Lift a block out of create_session.sh so the test runs the shipped code
+    rather than a paraphrase of it. A moved marker fails loudly here.
+
+    `close` closes the construct the end marker opens: the block runs to the
+    first line equal to it after the end marker (e.g. the `fi` of an `if`)."""
+    lines = open(path).read().splitlines(True)
+    start = next((i for i, l in enumerate(lines) if start_marker in l), None)
+    assert start is not None, "start marker %r is gone from %s" % (start_marker, path)
+    end = next((i for i, l in enumerate(lines[start:], start) if end_marker in l), None)
+    assert end is not None, "end marker %r is gone from %s" % (end_marker, path)
+    if close is not None:
+        end = next((i for i, l in enumerate(lines[end:], end) if l.strip() == close), None)
+        assert end is not None, "block opened at %r is never closed with %r" % (end_marker, close)
+    return "".join(lines[start:end + 1])
+
+
+def _seed_job(tmp_path, job_id, status="pending", started_token=None):
+    jobs = tmp_path / ".mam" / "jobs"
+    jobs.mkdir(parents=True, exist_ok=True)
+    (jobs / ("%s.json" % job_id)).write_text(json.dumps(
+        {"job_id": job_id, "status": status, "dispatch_token": "tok-1",
+         "attempt_id": 1, "ack_timeout_sec": 2}))
+    ev = jobs / ("%s.events.log" % job_id)
+    if started_token is None:
+        if ev.exists():
+            ev.unlink()
+    else:
+        ev.write_text(json.dumps(
+            {"job": job_id, "event": "started", "dispatch_token": started_token}) + "\n")
+    return jobs
+
+
+def _run_delivery(tmp_path, inject_rc, job_id="job1", **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)
+    pub_log = tmp_path / "published.txt"
+    script = '''
+set -euo pipefail
+source "%s"
+SESSION_NAME=test-sess
+DELEGATE_JOB_ID=%s
+DELEGATE_JOB_TOKEN=tok-1
+DELEGATE_JOB_ATTEMPT=1
+DELEGATE_JOB_ACK_S=2
+instructions="hello"
+inject_instructions() { return %d; }
+delegate_publish_event() { printf '%%s\\n' "$2" >> "%s"; }
+%s
+echo DELIVERED
+''' % (lib, job_id, inject_rc, pub_log, block)
+    env = dict(os.environ, MAM_JOBS_DIR=str(tmp_path / ".mam" / "jobs"))
+    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                         cwd=str(tmp_path), env=env)
+    published = pub_log.read_text().split() if pub_log.exists() else []
+    return res, published
+
+
+def test_s7_create_session_no_longer_publishes_started_for_the_agent(mam_sandbox):
+    """S7 (source): create_session used to publish `started` itself, so the
+    registry showed a started job even when the agent never saw the prompt.
+    With that line present no started-based verdict can mean anything."""
+    src = open(_create_sh(mam_sandbox)).read()
+    assert 'delegate_publish_event "$DELEGATE_JOB_ID" started' not in src
+    assert "delegate_wait_started" in src
+
+
+def test_s7b_delivery_succeeds_on_the_agents_own_started(mam_sandbox):
+    """S7b (real registry files, stubbed injection): a started line carrying
+    this dispatch token is the evidence."""
+    res, published = _run_delivery(mam_sandbox, 0, started_token="tok-1")
+    assert "DELIVERED" in res.stdout, res.stdout + res.stderr
+    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_s7d_no_started_within_t_is_a_delivery_failure(mam_sandbox):
+    """S7d: fail closed, and say so on the job."""
+    res, published = _run_delivery(mam_sandbox, 0)
+    assert res.returncode == 1, res.stdout + res.stderr
+    assert "DELIVERED" not in res.stdout
+    assert published == ["error"], published
+
+
+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
+
+
+def test_s8_rc4_is_a_warning_and_the_started_wait_decides(mam_sandbox):
+    """S8 (stubbed injection): rc 4 is "Enter not seen accepted", which is
+    exactly where a still-starting TUI lands - #23's false failure. It must no
+    longer abort; the started wait decides both ways."""
+    res, published = _run_delivery(mam_sandbox, 4, started_token="tok-1")
+    assert "DELIVERED" in res.stdout, res.stdout + res.stderr
+    assert "rc=4" in res.stderr, res.stderr
+    assert published == [], published
+
+    res, published = _run_delivery(mam_sandbox, 4)
+    assert res.returncode == 1, res.stdout + res.stderr
+    assert published == ["error"], published
+
+
+@pytest.mark.parametrize("rc", [1, 2, 3])
+def test_s8b_the_other_injection_failures_still_abort_at_once(mam_sandbox, rc):
+    """S8b: rc 1/2/3 say the pane never quiesced, a dialog was blocking, or the
+    text never appeared. No later `started` repairs those, so they stay hard
+    failures - only rc 4 was relaxed."""
+    res, published = _run_delivery(mam_sandbox, rc, started_token="tok-1")
+    assert res.returncode == 1, res.stdout + res.stderr
+    assert "DELIVERED" not in res.stdout
+    assert published == ["error"], published
+
+
+# --------------------------------------------------------------------------
+# item 4 — a rollback verifies the kill (#23)
+# --------------------------------------------------------------------------
+
+def _run_rollback(tmp_path, kill_rc, still_alive, row=True, job_id="job1"):
+    lib = tmp_path / ".agents" / "skills" / "lib.sh"
+    yaml_path = tmp_path / ".mam" / "agent-sessions.yaml"
+    if row:
+        yaml_path.write_text(
+            "herdr_sessions:\n"
+            "- name: test-sess\n"
+            "  status: running\n"
+            "  role: creator\n"
+            "  delegate_job_id: %s\n"
+            "  pane:\n"
+            "    index: 0\n"
+            "    pid: 111\n"
+            "    cmd: claude\n"
+            "    cwd: %s\n" % (job_id, tmp_path))
+    else:
+        yaml_path.write_text("herdr_sessions: []\n")
+    _seed_job(tmp_path, job_id)
+    block = _extract(_create_sh(tmp_path), "# _mark_row_rolled_back <status> <note>",
+                     "trap cleanup_herdr_on_error EXIT")
+    pub_log = tmp_path / "published.txt"
+    # The trap is installed and fired the way create_session fires it: an
+    # errexit shell exiting non-zero. rc 9 through means "no exit 43".
+    script = '''
+set -euo pipefail
+source "%s"
+SESSION_NAME=test-sess
+DELEGATE_JOB_ID=%s
+DELEGATE_JOB_TOKEN=tok-1
+DELEGATE_JOB_ATTEMPT=1
+ROW_APPENDED=%d
+AGENT_SESSIONS_YAML="%s"
+delegate_publish_event() { printf '%%s\\n' "$2" >> "%s"; }
+_herdr() {
+  case "${1:-}" in
+    kill-session) return %d ;;
+    has-session)  return %d ;;
+  esac
+  return 0
+}
+%s
+exit 9
+''' % (lib, job_id, 1 if row else 0, yaml_path, pub_log,
+       kill_rc, 0 if still_alive else 1, block)
+    env = dict(os.environ, MAM_JOBS_DIR=str(tmp_path / ".mam" / "jobs"),
+               AGENT_SESSIONS_YAML=str(yaml_path))
+    res = subprocess.run(["bash", "-c", script], capture_output=True, text=True,
+                         cwd=str(tmp_path), env=env)
+    published = pub_log.read_text().split() if pub_log.exists() else []
+    return res, published, yaml_path.read_text()
+
+
+def test_s9_a_kill_that_left_the_session_alive_is_reported(mam_sandbox):
+    """S9 (stubbed herdr, real YAML + registry): the kill failed and the
+    session is still up. #23's rollback swallowed exactly this and printed
+    "rolled back" anyway, leaving a `running` row against a failed job."""
+    res, published, yaml_text = _run_rollback(mam_sandbox, kill_rc=1, still_alive=True)
+    assert res.returncode == 43, res.stdout + res.stderr
+    assert "test-sess" in res.stderr and "kill-session" in res.stderr
+    assert "rollback-failed" in yaml_text, yaml_text
+    assert "status: running" in yaml_text, "a session that is still up must not be called gone"
+    assert "delegate_job_id: null" in yaml_text, yaml_text
+    assert published == ["cancelled"], published
+
+
+def test_s10_a_successful_kill_keeps_the_previous_behaviour(mam_sandbox):
+    """S10: nothing is still running, so no exit 43 and no warning - but the
+    row and the job are still cleaned up, which is the other half of #23."""
+    res, published, yaml_text = _run_rollback(mam_sandbox, kill_rc=0, still_alive=False)
+    assert res.returncode == 9, res.stdout + res.stderr
+    assert "rolled-back" in yaml_text, yaml_text + "\nSTDERR:" + res.stderr
+    assert "status: terminated" in yaml_text, yaml_text
+    assert "delegate_job_id: null" in yaml_text, yaml_text
+    assert published == ["cancelled"], published
+
+
+def test_s10b_a_rollback_before_the_row_existed_touches_no_yaml(mam_sandbox):
+    """S10b: failing before the append must not invent a row."""
+    res, published, yaml_text = _run_rollback(mam_sandbox, kill_rc=0, still_alive=False,
+                                              row=False)
+    assert res.returncode == 9, res.stdout + res.stderr
+    assert "herdr_sessions: []" in yaml_text, yaml_text