Plan 17 M4: migrate the harness and bundles onto the ledger

The fake daemon pushes 7.2 event batches (opening round: lifecycle,
turn text, tool call + result, usage; scripted corpora carry their own
kinds); the results channel drops its entries. Bundles assert the
events table, the JSON page and SSE stream, per-session seq
idempotency (redelivery stores nothing, a seq-less batch refuses
atomically), ledger-backed progress feeds, the goal narrative's
event projection, and the state-channel failure idempotency. goalui
migrates off the terminal page (session-row notices, the log page,
the runs-table link) and the yeehaw checks die with the feature.

test/bundles/allocation.py

@@ -505,14 +505,13 @@ def run(ctx):
         """Push-era raises on the dashboard fixture task (the poll-era
         /update channel is gone): a blocked results push carries the
         block payload; a review verb rides a review entry."""
-        if kind == "review" and summary is not None:
+        if kind == "review":
+            block = {"question": question, "type": "review"}
+            if summary is not None:
+                block["body"] = summary
             return saddle_api("POST", "/api/saddle/results", fake_token, {
-                "session_id": dash_session, "state": "working",
-                "entries": [{"id": 9317,
-                             "command": "saddle review",
-                             "output": summary, "type": "review",
-                             "timestamp": time.strftime(
-                                 "%Y-%m-%dT%H:%M:%SZ", time.gmtime())}]})[0]
+                "session_id": dash_session, "state": "blocked",
+                "block": block})[0]
         block = {"question": question, "type": kind}
         if body is not None:
             block["body"] = body
@@ -531,10 +530,17 @@ def run(ctx):
     conn.execute("insert into task_comments (task_id, author, body, created_at)"
                  " values (?, 'user:admin', 'timeline fixture comment by a human',"
                  " strftime('%s','now','-5 seconds'))", (cap_task,))
-    conn.execute("insert into task_progress (task_id, summary, created_at)"
-                 " values (?, 'timeline fixture progress',"
-                 " strftime('%s','now','-5 seconds'))",
-                 (cap_task,))
+    conn.execute(
+        "insert into events (kind, actor, scope_type, scope_id, payload,"
+        " created_at) values ('comment', 'user', 'task', ?,"
+        " ?, strftime('%s','now','-5 seconds'))",
+        (cap_task, json.dumps({"comment_id": 0, "author": "user:admin",
+                               "body": "timeline fixture comment by a human"})))
+    conn.execute(
+        "insert into events (kind, actor, scope_type, scope_id, payload,"
+        " created_at) values ('progress', 'model', 'task', ?,"
+        " ?, strftime('%s','now','-5 seconds'))",
+        (cap_task, json.dumps({"text": "timeline fixture progress"})))
     conn.commit()
     conn.close()
     # The probe RFC raised last: the newest pinned card and the newest
@@ -663,7 +669,7 @@ def run(ctx):
     conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at,"
                  " outcome) values (?, ?, strftime('%s','now'),"
                  " strftime('%s','now'), 'parked')",
-                 (plain_parked, fake_id))
+                 (plain_parked, racer2_card_id))
     conn.commit()
     conn.close()
     task_page = http_get_page(auth, "/tasks/%d" % plain_parked)
@@ -677,6 +683,10 @@ def run(ctx):
     detail = task_state(auth, plain_parked)
     check(detail.get("force_wake") is not None,
           "the task state exposes the offered force-wake action")
+    # Determinism: the wake's trigger-ask (and any idle-probe ask) must
+    # not race the MANUAL claim the section asserts on.
+    fake2.ask_on_wake = False
+    fake2.ask_on_idle_status = False
     act(auth, detail["force_wake"], {})
     status, resume_payload = fake2.ask_now()
     check(status == 200 and int(resume_payload["task"]["id"]) == plain_parked,
@@ -691,6 +701,8 @@ def run(ctx):
     conn.close()
     check(flag_row is not None and flag_row[0] in (0, False),
           "the claim consumed the task force-wake flag")
+    fake2.ask_on_wake = True
+    fake2.ask_on_idle_status = True
     # The woken run re-parks (fake daemon finishes then suspends? no —
     # fake.start_payloads script: the plain task runs to done via the
     # daemon's own finish path in later suites; here just let it be

test/bundles/dispatch.py

@@ -242,7 +242,7 @@ def run(ctx):
           "project_notes char cap: head kept, giant truncated (entries=%d, chars=%d)"
           % (len(notes), total))
 
-    print("== phase 2: pushed chunks -> raw log + terminal view ==")
+    print("== phase 2: pushed events -> the unified ledger ==")
     # Wait for the last dispatch to finish so its pushes landed.
     check(wait_task_done(auth, task_id_of("Char cap dispatch"), timeout=60) is not None,
           "char-cap task completes")
@@ -250,129 +250,110 @@ def run(ctx):
     raw_session = conn.execute(
         "select s.id from sessions s join tasks t on t.id = s.task_id"
         " where t.title = 'Char cap dispatch' order by s.id desc limit 1").fetchone()[0]
-    rows = conn.execute("select seq, chunk from session_raw_log where session_id = ?"
-                        " order by seq", (raw_session,)).fetchall()
+    rows = conn.execute(
+        "select seq, kind from events where session_id = ? order by seq",
+        (raw_session,)).fetchall()
     conn.close()
-    check(len(rows) >= 3, "session_raw_log captured the pushed chunks (%d)" % len(rows))
+    check(len(rows) >= 5,
+          "the ledger captured the round's event batch (%d)" % len(rows))
+    kinds = [r[1] for r in rows]
+    check(all(k in kinds for k in
+              ("lifecycle", "turn_text", "tool_call", "tool_result", "usage")),
+          "the opening batch carries the §7.2 kinds (%s)" % sorted(set(kinds)))
     seqs = [r[0] for r in rows]
     check(seqs == sorted(seqs) and len(set(seqs)) == len(seqs),
-          "pushed chunks stored once, seq strictly increasing")
-    text = "".join(r[1] for r in rows)
-    check(fake.PROMPT in text and "echo hello" in text and "hello" in text,
-          "pushed raw log keeps the prompt and echoed output byte-exact")
-    check("working-heartbeat" in text,
-          "the flush-timer heartbeat chunks keep arriving mid-round")
-
-    terminal = slot_public(auth, "/tasks/%d/terminal" % task_id_of("Char cap dispatch"),
-                           "terminal")
-    check(terminal.get("has_raw_log") is True and int(terminal["session_id"]) == raw_session,
-          "terminal view selects raw rendering for the raw session")
-
-    # Old sessions (typed entries, no raw rows) keep the fallback view
-    # (a dedicated fixture task: its only session is the legacy one,
-    # so the terminal view cannot prefer a dispatched raw session).
-    conn = db()
-    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
-                 " values (1, 'Legacy terminal task', 'working', 5, strftime('%s','now'))")
-    legacy_task = conn.execute("select max(id) from tasks").fetchone()[0]
-    conn.execute("insert into sessions (task_id, saddle_id, started_at, ended_at, outcome)"
-                 " values (?, ?, strftime('%s','now'), strftime('%s','now'), 'done')",
-                 (legacy_task, fake_id))
-    conn.commit()
-    legacy_session = conn.execute("select max(id) from sessions").fetchone()[0]
-    conn.execute("insert into session_commands (session_id, command_text, output_text,"
-                 " type, started_at) values (?, 'ls', 'old-output', 'command',"
-                 " strftime('%s','now'))", (legacy_session,))
-    conn.commit()
-    conn.close()
-    legacy = slot_public(auth, "/tasks/%d/terminal" % legacy_task, "terminal")
-    check(legacy.get("has_raw_log") is False,
-          "legacy session falls back to the entry-type terminal view")
-
-    request = urllib.request.Request(WEB + "/terminal-stream/%d" % raw_session)
-    with urllib.request.urlopen(request, timeout=15) as stream:
-        raw_events = []
-        deadline = time.monotonic() + 8
-        while time.monotonic() < deadline and len(raw_events) < 3:
-            line = stream.readline().decode().strip()
-            if not line.startswith("data: ") or line == "data: ping":
-                continue
-            event = json.loads(line[6:])
-            if event.get("type") == "raw":
-                raw_events.append(event)
-        check(len(raw_events) >= 3 and raw_events[0]["text"] == fake.PROMPT
-              and raw_events[0]["seq"] < raw_events[1]["seq"],
-              "terminal stream serves pushed chunks in seq order (prompt first)")
+          "stored events keep one strictly increasing seq per session")
+
+    # The plan-17 event log wire: one newest-first JSON page per task.
+    cap_task = task_id_of("Char cap dispatch")
+    status, page = saddle_api(
+        "GET", "/tasks/%d/events?limit=30" % cap_task, None)
+    check(status == 200 and page.get("ok") is True
+          and len(page.get("events", [])) >= 5,
+          "GET /tasks/<id>/events serves the newest-first page")
+    page_events = page["events"]
+    ids = [e["id"] for e in page_events]
+    check(ids == sorted(ids, reverse=True),
+          "the JSON page lists events newest first")
+    check(page_events[0]["kind"] == "state_change"
+          and page_events[0]["payload"].get("to") == "done",
+          "the newest event is the terminal state change")
+
+    # The SSE stream seeds the most-recent window then stays open.
+    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % cap_task)
+    stream = urllib.request.urlopen(request, timeout=15)
+    stream_events = []
+    deadline = time.monotonic() + 6
+    while time.monotonic() < deadline and len(stream_events) < 30:
+        line = stream.readline().decode().strip()
+        if not line.startswith("data: ") or line == "data: ping":
+            continue
+        event = json.loads(line[6:])
+        stream_events.append(event)
+    stream.close()
+    check(len(stream_events) >= 5,
+          "the event stream seeds the most-recent window (%d)" % len(stream_events))
+    stream_ids = [e["id"] for e in stream_events]
+    check(stream_ids == sorted(stream_ids) and len(set(stream_ids)) == len(stream_ids),
+          "the seed batch is oldest first inside the window")
 
     print("== phase 2: chunks/results ingestion — auth, idempotency ==")
     # Token auth on both push channels (plan 10: the per-saddle
     # bearer authorises every daemon->web call).
     status, _ = saddle_api("POST", "/api/saddle/chunks", None,
-                           {"session_id": raw_session, "chunks": []})
+                           {"session_id": raw_session, "events": []})
     check(status == 401, "chunks endpoint rejects a missing token")
     status, _ = saddle_api("POST", "/api/saddle/results", "saddle-bogus-token",
                            {"session_id": raw_session, "state": "working"})
     check(status == 401, "results endpoint rejects a wrong token")
     status, _ = saddle_api("POST", "/api/saddle/chunks", fake_token,
-                           {"session_id": 99999, "chunks": []})
+                           {"session_id": 99999, "events": []})
     check(status == 400, "chunks for an unknown session rejected")
     status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                            {"session_id": raw_session, "state": "done"})
     check(status == 200, "the owning saddle's session accepts its results")
 
-    # Pushed chunk idempotency: re-pushing a batch (the outbox's
+    # Pushed event idempotency: re-pushing a batch (the outbox's
     # at-least-once redelivery after a lost response) consumes the
     # duplicates and stores only the tail.
-    conn = db()
-    before = conn.execute("select count(*) from session_raw_log where session_id = ?",
-                          (raw_session,)).fetchone()[0]
-    conn.close()
-    redelivery = [{"seq": r[0], "text": r[1]} for r in rows[:3]]
-    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
-                              {"session_id": raw_session, "chunks": redelivery})
-    check(status == 200 and body.get("ok") is True and int(body.get("stored") or 0) == 0,
-          "a fully-duplicate batch stores nothing (duplicates consumed)")
-    fresh = [{"seq": 9000, "text": "late-flush\r\n"},
-             {"seq": 9001, "text": ""}]  # truly empty text: no row
-    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
-                              {"session_id": raw_session, "chunks": fresh})
-    check(status == 200 and int(body.get("stored") or 0) == 1,
-          "a partial-redelivery batch stores only the new chunks")
-    conn = db()
-    after = conn.execute("select count(*) from session_raw_log where session_id = ?",
-                         (raw_session,)).fetchone()[0]
-    empty_row = conn.execute(
-        "select count(*) from session_raw_log where session_id = ? and seq = 9001",
-        (raw_session,)).fetchone()[0]
-    conn.close()
-    check(after == before + 1 and empty_row == 0,
-          "redelivered chunks never duplicate rows; empty text stores nothing")
-
-    # Pushed entries: today's OutputEntry shape with the daemon's
-    # global entry ids — a redelivered results push stores its
-    # entries exactly once.
-    def command_rows(session):
+    def session_event_count(session):
         conn = db()
         try:
             return conn.execute(
-                "select command_text, type from session_commands"
-                " where session_id = ? order by id", (session,)).fetchall()
+                "select count(*) from events where session_id = ?",
+                (session,)).fetchone()[0]
         finally:
             conn.close()
 
-    entries = [{"id": 8101, "command": "sed -n 1p README.md", "output": "# demo",
-                "exit_code": 0, "type": "command", "timestamp": "2026-09-06T00:00:00Z"}]
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": raw_session, "state": "working",
-                            "entries": entries})
-    check(status == 200, "entries push accepted")
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": raw_session, "state": "working",
-                            "entries": entries})
-    check(status == 200, "redelivered entries push accepted (at-least-once)")
-    stored = command_rows(raw_session)
-    check(len([r for r in stored if r[0] == "sed -n 1p README.md"]) == 1,
-          "redelivered entries store exactly once (daemon-global dedup)")
+    before = session_event_count(raw_session)
+    redelivery = [{"seq": r[0], "kind": r[1]} for r in rows[:3]]
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": raw_session, "events": redelivery})
+    check(status == 200 and body.get("ok") is True and int(body.get("stored") or 0) == 0,
+          "a fully-duplicate batch stores nothing (duplicates consumed)")
+    fresh = [{"seq": 9000, "kind": "progress",
+              "payload": {"text": "late-flush\r\n"}},
+             {"seq": 9001, "kind": "progress", "payload": {}}]
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": raw_session, "events": fresh})
+    check(status == 200 and int(body.get("stored") or 0) == 2,
+          "a partial-redelivery batch stores only the new events")
+    check(session_event_count(raw_session) == before + 2,
+          "redelivered events never duplicate rows")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": raw_session,
+                               "events": [{"kind": "progress", "payload": {}}]})
+    check(status == 400, "an event without a positive seq is refused")
+
+    # A seq-less batch shape is the daemon contract violation the web
+    # rejects wholesale (nothing of the batch stores).
+    count_before = session_event_count(raw_session)
+    status, _ = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                           {"session_id": raw_session,
+                            "events": [{"seq": 9100, "kind": "progress"},
+                                       {"kind": "usage", "payload": {}}]})
+    check(status == 400 and session_event_count(raw_session) == count_before,
+          "a batch with a seq-less event is refused atomically")
 
     # The retired daemon routes 404 (phase 2 deleted /output, /log
     # and /start serving) — and the web app never calls them.

test/bundles/goals.py

@@ -423,13 +423,17 @@ def run(ctx):
         "select max(id) from sessions where task_id = ?",
         (run_task,)).fetchone()[0]
     conn.execute(
-        "insert into task_progress (task_id, summary, created_at)"
-        " values (?, 'Overseeing the flange fan-out', strftime('%s','now'))",
-        (run_task,))
+        "insert into events (session_id, seq, kind, actor, scope_type, scope_id,"
+        " payload, created_at) values (?, 8001, 'progress', 'model', 'task', ?,"
+        " ?, strftime('%s','now'))",
+        (open_run_session, run_task, json.dumps({"text": "Overseeing the flange fan-out"})))
     conn.execute(
-        "insert into session_commands (session_id, command_text, type, started_at)"
-        " values (?, 'git -C demo log --oneline -5', 'command',"
-        " strftime('%s','now'))", (open_run_session,))
+        "insert into events (session_id, seq, kind, actor, scope_type, scope_id,"
+        " payload, created_at) values (?, 8002, 'tool_call', 'model', 'task', ?,"
+        " ?, strftime('%s','now'))",
+        (open_run_session, run_task,
+         json.dumps({"tool": "run_command",
+                     "args": {"command": "git -C demo log --oneline -5"}})))
     conn.commit()
     conn.close()
     # A manual refresh records the daemon's health snapshot (the
@@ -439,15 +443,15 @@ def run(ctx):
     check(state["progress"] and state["progress"][0]["summary"]
           == "Overseeing the flange fan-out",
           "the goal page carries the agent's live progress feed")
-    check(state["current_command"] == "git -C demo log --oneline -5"
+    check(state["current_command"] == "run_command: git -C demo log --oneline -5"
           and state["current_command_running"] is True,
           "the current activity shows the open round's running command")
     check(state["agent_saddle"] == "fake-saddle"
           and "fake-saddle" in state["saddle_health"],
           "the assigned agent saddle and its health display")
     check(state["runs"] and state["runs"][0]["saddle_name"] == "fake-saddle"
-          and "session_id=" in state["runs"][0]["terminal_url"],
-          "the run table carries host saddles and deep-linked terminals")
+          and state["runs"][0]["terminal_url"] == "/tasks/%d/log" % run_task,
+          "the run table carries host saddles and the event log link")
     goal_page = http_get_page(auth, "/goals/%d" % goal1)
     check('stm-markdown="goal.brief"' in goal_page
           and 'stm-markdown="goal.plan"' in goal_page
@@ -1023,20 +1027,43 @@ def run(ctx):
     print("== goal 4 t1: force wake — the human resume for a parked agent ==")
     # The goal-2 stall class: the parked run has NO pending wake (a
     # draft task never enqueued → no terminal transition → parked
-    # forever). Force wake is the human recovery lever.
+    # forever). Force wake is the human recovery lever. A fresh goal
+    # owns the scenario (goal1's run completed with its goal).
+    stall_goal = create_goal({"project_id": "1", "title": "Stall goal",
+                              "brief": "Parks with a never-enqueued draft.",
+                              "branch_name": "goal/stall"})
+    fake.goal_script = [("plan", "# Stall plan\n\n1. Wait for the stall")]
+    act(auth, goal_state(stall_goal)["start"], {})
+    check(wait_for(lambda: goal_state(stall_goal)["status"] == "plan_review" or False,
+                   timeout=60, what="stall plan enters review") is not None,
+          "the stall goal reaches plan review")
+    act(auth, goal_state(stall_goal)["plan_pending_block"]["answer_action"],
+        {"verdict": "accept"})
+    check(wait_for(lambda: goal_state(stall_goal)["status"] == "active" or False,
+                   timeout=30, what="stall goal activates") is not None,
+          "the stall goal activates")
     fake.goal_script = [("suspend",)]
+    conn = db()
+    run_task = conn.execute(
+        "select id from tasks where goal_id = ? and kind = 'goal_run'",
+        (stall_goal,)).fetchone()[0]
+    conn.close()
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                    or False, timeout=60, what="agent parks for force wake") is not None,
           "the goal agent parks before the force-wake scenario")
     # The stall fixture: an agent-created draft that was never enqueued
     # (typo'd enqueue) — statuses must name it as NOT enqueued.
+    conn = db()
+    agent_session = conn.execute(
+        "select max(id) from sessions where task_id = ?", (run_task,)).fetchone()[0]
+    conn.close()
     status, body = saddle_api("POST", "/api/saddle/task/create", fake_token, {
         "session_id": agent_session, "title": "QA sweep",
         "mandate": "Run the QA sweep.", "mode": "test"})
     check(status == 200 and body.get("ok") is True,
           "the stall fixture draft task creates (%s)" % body.get("error"))
     stall_draft = int(body["task_id"])
-    goal_page = http_get_page(auth, "/goals/%d" % goal1)
+    goal_page = http_get_page(auth, "/goals/%d" % stall_goal)
     check('stm-action="goal.force_wake"' in goal_page,
           "the goal page authors the Force wake button while the run is parked")
     check('stm-if="goal.force_wake != null"' in goal_page,
@@ -1047,7 +1074,7 @@ def run(ctx):
                        lambda: len(fake.goal_assignments) == parked_runs,
                        "no press, no wake"),
           "without a press the parked agent never wakes")
-    state = goal_state(goal1)
+    state = goal_state(stall_goal)
     check(state.get("force_wake") is not None,
           "the goal state exposes the offered force-wake action")
     act(auth, state["force_wake"], {})
@@ -1073,32 +1100,32 @@ def run(ctx):
                        "consumed flag never re-fires"),
           "the consumed flag does not re-wake (one wake per press)")
     # Guards: a paused goal refuses the wake (Resume is its lever).
-    state = goal_state(goal1)
+    state = goal_state(stall_goal)
     act(auth, state["pause"], {})
     conn = db()
     paused_flag = conn.execute("select force_wake from goals where id = ?",
-                               (goal1,)).fetchone()
+                               (stall_goal,)).fetchone()
     conn.close()
     check(paused_flag is not None and paused_flag[0] in (0, False),
           "pausing does not set the flag")
-    state = goal_state(goal1)
+    state = goal_state(stall_goal)
     check(state.get("force_wake") is None,
           "a paused goal does not author the Force wake action")
     act(auth, state["resume"], {})
-    check(wait_for(lambda: goal_state(goal1)["status"] == "active" or False,
+    check(wait_for(lambda: goal_state(stall_goal)["status"] == "active" or False,
                    timeout=30, what="goal reactivates") is not None,
           "the goal reactivates for the guard checks")
     # Non-suspended run refuses: run is parked again (script parked it
     # above), wake it, let it park, then a second press while parked is
     # fine — the 400 path needs a NON-suspended run. Start there: wake
     # once more and press while it is working.
-    state = goal_state(goal1)
+    state = goal_state(stall_goal)
     act(auth, state["force_wake"], {})
     check(wait_for(lambda: len(fake.goal_assignments) > woke_runs or False,
                    timeout=30, what="second wake fires") is not None,
           "a second press wakes again (re-press is allowed)")
     # While it is mid-script (not yet parked), the action is hidden.
-    check(goal_state(goal1).get("force_wake") is None
+    check(goal_state(stall_goal).get("force_wake") is None
           or task_state(auth, run_task)["status"] != "suspended",
           "the wake action hides while the run is not parked")
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"

test/bundles/goalui.py

@@ -9,7 +9,8 @@ def run(ctx):
     B1 — goal-run control parity (Retry supervision / Requeue-to-resume
     on the status head, the /tasks/<goal-run-id> redirect, no
     notify-me toggle), B2 — the plan-review remarks queue, B4 —
-    sessions.resume_notice + the terminal dropdown/resume panel, B5 —
+    sessions.resume_notice on the task page's session rows + the
+    plan-17 event log page, B5 —
     page order, effective rules and per-task capability badges, B6 —
     narrative footnote weights. C1 — the single-row block world: the
     goal's blocks ARE the run task's rows (answered from every
@@ -106,8 +107,8 @@ def run(ctx):
     check("Retry supervision" in goal_page and "Requeue to resume" in goal_page,
           "the status head ships the run-control parity buttons")
     check("status-corner" not in goal_page and "goal complete" not in goal_page
-          and goal_page.count("badge-yeehaw") == 1,
-          "yeehaw is rows-only (no goal-head corner badge; the row badge remains)")
+          and "badge-yeehaw" not in goal_page,
+          "no yeehaw badge remains anywhere (the feature died with the entry stream)")
 
     check(goal_state(goal)["current_activity"] == "parked awaiting the plan review",
           "the head line reads 'parked awaiting the plan review' while one is open")
@@ -261,20 +262,15 @@ def run(ctx):
     check(not goal_side_tables,
           "no goal-side store exists — historical stragglers are impossible")
 
-    print("== B4: the terminal session dropdown + resume-info panel ==")
+    print("== B4: resume notices on the session rows + the event log page ==")
     fake.goal_script = [("suspend",)]
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                    or False, timeout=60, what="agent parks suspended") is not None,
           "the agent parks suspended for the requeue scenario")
-    terminal = slot_public(auth, "/tasks/%d/terminal" % run_task, "terminal")
-    ids = [int(s["id"]) for s in terminal["sessions"]]
+    sessions = task_state(auth, run_task)["sessions"]
+    ids = [int(s["id"]) for s in sessions]
     check(ids == sorted(ids, reverse=True),
-          "the session dropdown lists newest-first")
-    check(int(terminal["session_id"]) == ids[0],
-          "the terminal still defaults to the latest session")
-    labels = {int(s["id"]): s["label"] for s in terminal["sessions"]}
-    check(all(label.startswith("Session #") for label in labels.values()),
-          "dropdown labels read 'Session #N — started <time>'")
+          "the task page's session history lists newest-first")
     # Recompute at the check point: the C1 answers above each
     # resumed the agent too, so the earlier snapshot would go stale.
     conn = db()
@@ -282,26 +278,21 @@ def run(ctx):
         "select id, resume_notice from sessions where task_id = ?",
         (run_task,)).fetchall() if text]
     conn.close()
-    resumed_markers = [i for i in ids if "[resumed]" in labels[i]]
-    check(set(resumed_markers) == set(resumed_ids),
-          "the [resumed] marker rides exactly the notice-carrying sessions")
-    notices_by_id = {int(s["id"]): s.get("notice") or "" for s in terminal["sessions"]}
-    check("Opening brief" in notices_by_id[min(ids)]
-          and "Drive the plan-14 goal surface" in notices_by_id[min(ids)],
-          "session 1's panel shows the opening-brief framing (from the goal brief)")
-    check(any("Plan feedback received" in (notices_by_id[i] or "") for i in resumed_ids),
-          "a resumed session's panel shows the stored resume notice")
-    terminal_page = http_get_page(auth, "/tasks/%d/terminal" % run_task)
-    check('id="session-select"' in terminal_page
-          and "window.terminalPickSession" in terminal_page,
-          "the session picker is an options dropdown")
-    check('stm-markdown="window.terminalSessionNotice(terminal)"' in terminal_page,
-          "the panel renders the selected session's notice as markdown")
-    check("No opening brief or resume notice recorded" in terminal_page,
-          "the panel ships its graceful empty state")
-    css = open(os.path.join(REPO, "src/pages/app.css")).read()
-    check("resume-notice-panel" in css and "max-height: 200px" in css,
-          "the panel is a 200px vertical-scroll strip (stylesheet)")
+    notices_by_id = {int(s["id"]): s.get("notice") or "" for s in sessions}
+    check(set(i for i in ids if notices_by_id[i]) == set(resumed_ids),
+          "the session rows' notice column rides exactly the notice-carrying sessions")
+    check(any("Plan feedback received" in notices_by_id[i] for i in resumed_ids),
+          "a resumed session's row shows the stored resume notice")
+    log_state = slot_public(auth, "/tasks/%d/log" % run_task, "task_log")
+    check(log_state.get("exists") is True and int(log_state["task_id"]) == run_task
+          and log_state["task_url"] == "/tasks/%d" % run_task,
+          "the event log page hydrates the run-task header")
+    log_page = http_get_page(auth, "/tasks/%d/log" % run_task)
+    check("EventSource" in log_page and "events/stream" in log_page
+          and "data-log-feed" in log_page,
+          "the log page subscribes to the ledger's SSE stream")
+    check(goal_state(goal)["runs"][0]["terminal_url"] == "/tasks/%d/log" % run_task,
+          "the goal runs table links the event log")
 
     print("== B1: requeue-to-resume from the goal head ==")
     conn = db()
@@ -490,12 +481,12 @@ def run(ctx):
           "the task rows render capability badges")
 
     print("== B6: narrative weights — events full, comments footnote ==")
-    # An AGENT comment projects the narrative's comment event (human
-    # comments only thread) — pushed through the entries channel while
+    # An AGENT remark rides the ledger as a comment event (human
+    # comments only thread) — pushed through the events channel while
     # parked, so the suspended state stays put.
-    fake.push_results("suspended", entries=[{
-        "id": 9001, "command": "saddle comment", "output": "An agent aside.",
-        "exit_code": 0, "type": "comment", "timestamp": "2026-09-12T00:00:00Z"}])
+    fake.push_events([{"kind": "comment", "actor": "model",
+                       "payload": {"author": "agent",
+                                   "body": "An agent aside."}}])
     check(wait_for(lambda: any(e["kind"] == "comment"
                                for e in goal_state(goal)["events"]) or False,
                    timeout=15, what="comment event lands") is not None,
@@ -531,31 +522,24 @@ def run(ctx):
           "the task page renders system rows with the same style")
 
     print("== S5 U2: mechanical rows filtered from user-facing timelines ==")
-    # The daemon's mechanical system entries (the park notice, the
-    # restore footnote) ride the wire typed 'system': the raw terminal
-    # row STORES (session_commands), but no comment row, narrative
-    # event or dashboard timeline row is authored for them.
+    # The daemon's mechanical system remarks (the park notice, the
+    # restore footnote) ride the ledger as harness comment events: the
+    # event row STORES, but no comment row, narrative event or
+    # dashboard timeline row renders for them.
     open_run = open_session(run_task)
     mechanical = [
-        {"id": 9501, "command": "saddle suspend",
-         "output": "goal agent parked until an owned task completes or"
-                   " fails; resume via /start resume_notice",
-         "exit_code": 0, "type": "system", "timestamp": "2026-09-12T00:00:00Z"},
-        {"id": 9502, "command": "harness resume",
-         "output": "[harness] session context restored — task resumed",
-         "exit_code": 0, "type": "system", "timestamp": "2026-09-12T00:00:01Z"},
+        {"seq": 9501, "kind": "comment", "actor": "harness",
+         "payload": {"author": "system",
+                     "body": "goal agent parked until an owned task"
+                             " completes or fails"}},
+        {"seq": 9502, "kind": "comment", "actor": "harness",
+         "payload": {"author": "system",
+                     "body": "[harness] session context restored — task resumed"}},
     ]
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": open_run, "state": "suspended",
-                            "entries": mechanical})
-    check(status == 200, "the mechanical system entries push cleanly")
-    conn = db()
-    raw_rows = conn.execute(
-        "select count(*) from session_commands where session_id = ?"
-        " and type = 'system'", (open_run,)).fetchone()[0]
-    conn.close()
-    check(raw_rows >= 2,
-          "the raw terminal keeps the mechanical rows (typed entries store)")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": open_run, "events": mechanical})
+    check(status == 200 and int(body.get("stored") or 0) == 2,
+          "the mechanical harness remarks store on the ledger")
     check(all("parked until" not in c["body"]
               and "session context restored" not in c["body"]
               for c in task_state(auth, run_task)["comments"]),
@@ -619,23 +603,20 @@ def run(ctx):
     check("Accept goal" in goal_page and "Reject with feedback" in goal_page
           and 'name="feedback"' in goal_page,
           "the completion review presents explicit accept/reject + free-text feedback (U1)")
-    # The burst shape: the done ENTRY and the done STATE in one push
-    # (fresh entry ids), then the whole pair redelivered — the
-    # pending-report idempotency leaves the one row alone.
+    # The burst shape: the done STATE redelivered under the run's
+    # newest session (the entries channel is gone — the report rides
+    # the summary) — the pending-report idempotency leaves the one
+    # row alone.
     conn = db()
     burst_session = conn.execute(
         "select max(id) from sessions where task_id = ?",
         (run_task,)).fetchone()[0]
     conn.close()
-    done_pair = [{"id": 9401, "command": "saddle done",
-                  "output": "# Completion\n\nOne block only.", "exit_code": 0,
-                  "type": "done", "timestamp": "2026-09-12T00:00:00Z"}]
     status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                            {"session_id": burst_session, "state": "done",
-                            "entries": done_pair,
                             "summary": "# Completion\n\nOne block only."})
     check(status == 200 and completion_block_count() == 1,
-          "the done entry + state pair raises nothing more (one block, ever)")
+          "a redelivered done push raises nothing more (one block, ever)")
     state = goal_state(goal)
     act(auth, state["completion_pending_block"]["answer_action"],
         {"verdict": "accept"})

test/bundles/lifecycle.py

@@ -378,9 +378,9 @@ def run(ctx):
     check(wait_task_done(auth, during_fence, timeout=60) is not None,
           "queued work dispatches once the saddle proves it is alive")
 
-    print("== R7 B3: health panel + pushed raw log from the first round ==")
-    # A normally dispatched task pushes its chunks immediately: raw
-    # rows exist from the first round (the session-4 gap stays
+    print("== R7 B3: health panel + pushed events from the first round ==")
+    # A normally dispatched task pushes its event batch immediately:
+    # ledger rows exist from the first round (the session-4 gap stays
     # fixed without any probe machinery).
     task_id_tmp = create_queued(auth, {
         "project_id": "1",
@@ -389,18 +389,18 @@ def run(ctx):
     })
     raw_first = task_id_tmp
 
-    def raw_count(task_id):
+    def event_count(task_id):
         conn = db()
         try:
             return conn.execute(
-                "select count(*) from session_raw_log where session_id in"
+                "select count(*) from events where session_id in"
                 " (select id from sessions where task_id = ?)", (task_id,)).fetchone()[0]
         finally:
             conn.close()
 
-    check(wait_for(lambda: raw_count(raw_first) > 0 or False,
-                   timeout=60, what="raw rows from first push") is not None,
-          "pushed chunks land from the first round (session-4 gap)")
+    check(wait_for(lambda: event_count(raw_first) > 0 or False,
+                   timeout=60, what="event rows from first push") is not None,
+          "pushed events land from the first round (session-4 gap)")
     check(wait_task_done(auth, raw_first, timeout=60) is not None,
           "raw-first task completes")
 
@@ -421,12 +421,14 @@ def run(ctx):
     # redelivery of the whole batch (the outbox's at-least-once
     # retry) consumes the duplicates.
     corpus = [
-        {"seq": 1, "text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "},
-        {"seq": 2, "text": "echo hello\r\n"},
-        {"seq": 3, "text": "\r\n"},
-        {"seq": 4, "text": "hello\r\n"},
-        {"seq": 5, "text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "},
-        {"seq": 6, "text": "done\r\n"},
+        {"seq": 1, "kind": "command_output",
+         "payload": {"text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "}},
+        {"seq": 2, "kind": "command_output", "payload": {"text": "echo hello\r\n"}},
+        {"seq": 3, "kind": "command_output", "payload": {"text": "\r\n"}},
+        {"seq": 4, "kind": "command_output", "payload": {"text": "hello\r\n"}},
+        {"seq": 5, "kind": "command_output",
+         "payload": {"text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "}},
+        {"seq": 6, "kind": "command_output", "payload": {"text": "done\r\n"}},
     ]
     fake.scripted_corpus = corpus
     fake.hold_state = True
@@ -441,9 +443,9 @@ def run(ctx):
         conn = db()
         try:
             return [r[0] for r in conn.execute(
-                "select seq from session_raw_log where session_id in"
+                "select seq from events where session_id in"
                 " (select id from sessions where task_id = ?)"
-                " order by seq", (task_id,))]
+                " and seq is not null order by seq", (task_id,))]
         finally:
             conn.close()
 
@@ -452,16 +454,15 @@ def run(ctx):
           "lossless task dispatches and pushes its opening flush")
     check(wait_for(lambda: stored_seqs(lossless) == [1, 2, 3, 4, 5, 6] or False,
                    timeout=30, what="all six chunks stored") is not None,
-          "every pushed chunk lands in session_raw_log, CRLF-only included (%s)"
+          "every pushed event lands in the ledger, CRLF-only included (%s)"
           % stored_seqs(lossless))
     status, _ = saddle_api("POST", "/api/saddle/chunks", fake.token,
                            {"session_id": fake.session_id,
-                            "chunks": [dict(chunk) for chunk in corpus]})
+                            "events": [dict(chunk) for chunk in corpus]})
     check(status == 200 and stored_seqs(lossless) == [1, 2, 3, 4, 5, 6],
           "a full-batch redelivery stores nothing new (exactly-once)")
     request = urllib.request.Request(
-        WEB + "/terminal-stream/%d" % [r[0] for r in db().execute(
-            "select id from sessions where task_id = ?", (lossless,))][0])
+        WEB + "/tasks/%d/events/stream" % lossless)
     with urllib.request.urlopen(request, timeout=15) as stream:
         texts = []
         deadline = time.monotonic() + 8
@@ -470,10 +471,10 @@ def run(ctx):
             if not line.startswith("data: ") or line == "data: ping":
                 continue
             event = json.loads(line[6:])
-            if event.get("type") == "raw":
-                texts.append(event.get("text") or "")
+            if event.get("kind") == "command_output":
+                texts.append(event.get("payload", {}).get("text") or "")
         check("\r\n" in texts and len(texts) == 6,
-              "SSE serves the whitespace chunk unfiltered")
+              "SSE serves the whitespace payload unfiltered")
     fake.hold_state = False
     fake.scripted_corpus = None
     check(wait_task_done(auth, lossless, timeout=60) is not None,

test/bundles/pages.py

@@ -65,9 +65,9 @@ def run(ctx):
           "task page carries the quote-comment dialog the popover opens")
     check('name="quote"' in page and 'name="return_to"' in page,
           "quote dialog posts the quote and return-to fields")
-    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
-    check(status == 200 and "quote-comment-dialog" in page,
-          "terminal page keeps its quote dialog")
+    status, page = http("GET", "/tasks/%d/log" % plan_task, timeout=15)
+    check(status == 200 and "data-log-feed" in page and "data-log-more" in page,
+          "the event log page renders the feed and the load-more control")
 
     print("== B3: review section hides the duplicate body ==")
     status, _ = raise_block(session_id, "Review the diff", kind="review",
@@ -252,45 +252,46 @@ def run(ctx):
     print("== rejected entries (non-fatal) ==")
     # The push wire: a rejected entry rides a round-results push, stores
     # on the session and streams through SSE; the task is untouched.
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token, {
-        "session_id": session_id, "state": "working",
-        "entries": [{
-            "id": 9021, "command": "bash -n script.sh",
-            "output": "syntax error near line 3 (recoverable)",
-            "exit_code": 1, "type": "rejected",
-            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
-        }]})
-    check(status == 200, "a rejected entry rides the results push")
-    request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token, {
+        "session_id": session_id,
+        "events": [{"seq": 9021, "kind": "tool_result", "actor": "harness",
+                    "payload": {"command": "bash -n script.sh",
+                                "stderr": "syntax error near line 3 (recoverable)",
+                                "exit_code": 1}}]})
+    check(status == 200 and int(body.get("stored") or 0) == 1,
+          "a failed tool result rides the events push")
+    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % plan_task)
     with urllib.request.urlopen(request, timeout=15) as stream:
-        rejected_event = None
+        result_event = None
         deadline = time.monotonic() + 8
-        while time.monotonic() < deadline and rejected_event is None:
+        while time.monotonic() < deadline and result_event is None:
             line = stream.readline().decode().strip()
             if not line.startswith("data: ") or line == "data: ping":
                 continue
             event = json.loads(line[6:])
-            if event.get("type") == "rejected":
-                rejected_event = event
-        check(rejected_event is not None
-              and "syntax error near line 3" in (rejected_event.get("output") or ""),
-              "terminal stream carries rejected entries with their reason")
-
-    # The page template renders them with the distinct non-fatal style.
-    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
-    check(status == 200 and "term-rejected" in page,
-          "terminal page renders rejected entries in their own style")
-
-    print("== B8: ANSI SGR in the terminal view ==")
-    # Byte-exact SGR delivery: the daemon's coloured chunks flow through
-    # the store + SSE untouched (the client renderer styles them).
+            if (event.get("kind") == "tool_result"
+                    and "syntax error near line 3" in
+                    (event.get("payload", {}).get("stderr") or "")):
+                result_event = event
+        check(result_event is not None,
+              "the event stream carries tool results with their reason")
+
+    # The page template renders the feed and the filters.
+    status, page = http("GET", "/tasks/%d/log" % plan_task, timeout=15)
+    check(status == 200 and "data-log-filter" in page and "data-log-search" in page,
+          "the event log page renders filters and search")
+
+    print("== B8: byte-exact payload delivery over the event stream ==")
+    # Byte-exact delivery: ANSI-coloured turn text flows through the
+    # ledger + SSE untouched (the client renderer styles them).
     coloured = "\x1b[1;32mbold green\x1b[0m then \x1b[38;5;196m256-red\x1b[0m\n"
-    conn = db()
-    conn.execute("insert into session_raw_log (session_id, seq, chunk)"
-                 " values (?, 900, ?)", (session_id, coloured))
-    conn.commit()
-    conn.close()
-    request = urllib.request.Request(WEB + "/terminal-stream/%d" % session_id)
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token, {
+        "session_id": session_id,
+        "events": [{"seq": 9100, "kind": "turn_text",
+                    "payload": {"turn": 9, "text": coloured}}]})
+    check(status == 200 and int(body.get("stored") or 0) == 1,
+          "the coloured turn text stores as one event")
+    request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % plan_task)
     with urllib.request.urlopen(request, timeout=15) as stream:
         coloured_event = None
         deadline = time.monotonic() + 8
@@ -299,14 +300,12 @@ def run(ctx):
             if not line.startswith("data: ") or line == "data: ping":
                 continue
             event = json.loads(line[6:])
-            if event.get("type") == "raw" and "\x1b[1;32m" in (event.get("text") or ""):
+            if (event.get("kind") == "turn_text"
+                    and "\x1b[1;32m" in (event.get("payload", {}).get("text") or "")):
                 coloured_event = event
         check(coloured_event is not None
-              and "\x1b[38;5;196m" in coloured_event["text"],
-              "SGR sequences stream byte-exact to the client renderer")
-    status, page = http("GET", "/tasks/%d/terminal" % plan_task, timeout=15)
-    check(status == 200 and "makeAnsiRenderer" in page and "colour256" in page,
-          "terminal page embeds the ANSI renderer (SGR + 256-colour palette)")
+              and "\x1b[38;5;196m" in coloured_event["payload"]["text"],
+              "payload bytes stream byte-exact to the client renderer")
 
     # Dispatch path: a rejected entry in a round's results push flows
     # into the audit trail without failing (or finishing) the task.
@@ -323,13 +322,10 @@ def run(ctx):
                       "max_tokens exceeded (recoverable)")
     check(wait_task_done(auth, rejected_task, timeout=60) is not None,
           "task completes normally past rejected entries")
-    conn = db()
-    stored = conn.execute(
-        "select count(*) from session_commands sc join sessions s on s.id = sc.session_id"
-        " where s.task_id = ? and sc.type = 'rejected'", (rejected_task,)).fetchone()[0]
-    conn.close()
-    check(stored >= 1,
-          "rejected entries stored in the audit trail (%d rows)" % stored)
+    stored = [row for row in event_rows(rejected_task, ["command_output"])
+              if "max_tokens exceeded" in row[1]]
+    check(len(stored) >= 1,
+          "command_output events stored in the audit trail (%d rows)" % len(stored))
     check(task_state(auth, rejected_task)["status"] == "done"
           and "fake daemon finished" in (task_state(auth, rejected_task).get("summary") or ""),
           "rejected entries never fail the task")
@@ -363,12 +359,13 @@ def run(ctx):
     layout_task = conn.execute("select max(id) from tasks").fetchone()[0]
     # backdated so the dashboard-timeline fixtures below stay the
     # newest events
-    conn.execute("insert into task_progress (task_id, summary, created_at)"
-                 " values (?, 'first update', strftime('%s','now','-90 seconds'))",
-                 (layout_task,))
-    conn.execute("insert into task_progress (task_id, summary, created_at)"
-                 " values (?, 'second update', strftime('%s','now','-89 seconds'))",
-                 (layout_task,))
+    for offset, line in ((90, "first update"), (89, "second update")):
+        payload = json.dumps({"text": line})
+        conn.execute(
+            "insert into events (kind, actor, scope_type, scope_id, payload,"
+            " created_at) values ('progress', 'model', 'task', ?, ?,"
+            " strftime('%s','now', ?))",
+            (layout_task, payload, "-%d seconds" % offset))
     conn.commit()
     conn.close()
     status, history_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
@@ -411,6 +408,21 @@ def run(ctx):
                  " strftime('%s','now'), strftime('%s','now'),"
                  " strftime('%s','now'))")
     report_empty = conn.execute("select max(id) from tasks").fetchone()[0]
+    # The terminal transitions ride the ledger (state_change events
+    # carrying the summary the timeline snippets).
+    for task_id, to, summary in (
+            (report_done, "done", report_md),
+            (report_failed, "failed", "Boom: tests failed\n\nsecond line carries the detail"),
+            (report_empty, "done", "")):
+        payload = {"from": "working", "to": to,
+                   "text": "Task completed" if to == "done" else "Task failed"}
+        if summary:
+            payload["summary"] = summary
+        conn.execute(
+            "insert into events (kind, actor, scope_type, scope_id, payload,"
+            " created_at) values ('state_change', 'system', 'task', ?, ?,"
+            " strftime('%s','now'))",
+            (task_id, json.dumps(payload)))
     conn.commit()
     conn.close()
 
@@ -544,130 +556,6 @@ def run(ctx):
           "the create expander offers the mode + auto-enqueue-after selects"
           " (no subtask checkboxes)")
 
-    print("== yeehaw: ingestion, festive card, badge, timeline ==")
-    # `saddle yeehaw <message>` arrives as a round-results entry with the
-    # message in the output field: it stores like any entry type, earns
-    # the task its 🤠 badge (derived, nothing new stored), renders as the
-    # task page's festive card (last 3, newest first, "+N more") and
-    # surfaces on the dashboard timeline as its own event kind.
-    conn = db()
-    conn.execute("insert into tasks (project_id, title, status, priority, created_at)"
-                 " values (1, 'Yeehaw task', 'working', 5, strftime('%s','now'))")
-    yeehaw_task = conn.execute("select max(id) from tasks").fetchone()[0]
-    conn.execute("insert into sessions (task_id, saddle_id, started_at)"
-                 " values (?, ?, strftime('%s','now'))", (yeehaw_task, fake_id))
-    yeehaw_session = conn.execute("select max(id) from sessions").fetchone()[0]
-    conn.commit()
-    conn.close()
-    now_epoch = int(time.time())
-
-    def push_yeehaw(entry_id, message, offset=0):
-        """One celebration riding a round-results push, timestamped
-        `offset` seconds from now (distinct epochs keep the timeline's
-        newest-first ordering observable)."""
-        return saddle_api("POST", "/api/saddle/results", fake_token, {
-            "session_id": yeehaw_session, "state": "working",
-            "entries": [{
-                "id": entry_id, "command": "saddle yeehaw", "output": message,
-                "exit_code": 0, "type": "yeehaw",
-                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ",
-                                           time.gmtime(now_epoch + offset)),
-            }]})[0]
-
-    def yeehaw_rows():
-        conn = db()
-        try:
-            return conn.execute(
-                "select count(*) from session_commands where session_id = ?"
-                " and type = 'yeehaw'", (yeehaw_session,)).fetchone()[0]
-        finally:
-            conn.close()
-
-    check(push_yeehaw(9101, "Shipped the flange") == 200,
-          "yeehaw entries ingest through the results push like any entry type")
-    check(yeehaw_rows() == 1,
-          "the yeehaw entry stores as a session_commands row (message in output)")
-    check(push_yeehaw(9101, "Shipped the flange") == 200 and yeehaw_rows() == 1,
-          "a redelivered yeehaw push is idempotent on the daemon entry id")
-    check(task_state(auth, yeehaw_task)["status"] == "working",
-          "a yeehaw entry never derails the task state")
-
-    push_yeehaw(9102, "Second hoedown", 1)
-    push_yeehaw(9103, "Third hoedown\nsecond line carries the detail", 2)
-    push_yeehaw(9104, "Fourth hoedown", 3)
-    push_yeehaw(9105, "Fifth hoedown", 4)
-    detail = task_state(auth, yeehaw_task)
-    check(detail.get("yeehaw") is True and detail.get("yeehaw_total") == 5,
-          "the detail state flags the badge presence and the uncapped total")
-    check([y["message"] for y in detail.get("yeehaws", [])]
-          == ["Fifth hoedown", "Fourth hoedown",
-              "Third hoedown\nsecond line carries the detail"],
-          "the festive card keeps the last 3 yeehaws, newest first (full messages)")
-    check(detail.get("yeehaw_more") == 2,
-          "the '+N more' hint counts the capped tail")
-    status, page = http("GET", "/tasks/%d" % yeehaw_task, timeout=15)
-    # The template compiler HTML-escapes the emoji (&#129312; = 🤠).
-    check(status == 200 and "yeehaw-card" in page and "&#129312; Yeehaw!" in page,
-          "the task page renders the festive yeehaw card")
-    check(page.index("<h3>Progress</h3>") < page.index("yeehaw-card")
-          < page.index("status-head"),
-          "the card sits between the progress feed and the status grid")
-    check('stm-text="task.yeehaw_more"' in page and "more yeehaw" in page,
-          "the card carries the +N more hint")
-    check("badge-yeehaw" in page and "status-corner" in page,
-          "the status card corner carries the earned 🤠 badge")
-    plain_detail = task_state(auth, layout_task)
-    check(plain_detail.get("yeehaw") is False and plain_detail.get("yeehaw_total") == 0
-          and plain_detail.get("yeehaws", []) == [],
-          "tasks without yeehaw entries carry no card state")
-    status, plain_page = http("GET", "/tasks/%d" % layout_task, timeout=15)
-    check(status == 200 and 'section class="card yeehaw-card" stm-if="task.yeehaw"' in plain_page,
-          "the card template only renders under state gating (stm-if task.yeehaw)")
-
-    queue_rows = slot_public(auth, "/tasks", "tasks")["tasks"]
-    check([t for t in queue_rows if int(t["id"]) == yeehaw_task][0].get("has_yeehaw") is True,
-          "queue rows badge yeehaw tasks")
-    check([t for t in queue_rows if int(t["id"]) == layout_task][0].get("has_yeehaw") is False,
-          "queue rows leave yeehaw-less tasks bare")
-    status, queue_page = http("GET", "/tasks", timeout=15)
-    check(status == 200 and 'stm-if="task.has_yeehaw"' in queue_page,
-          "the queue template renders the row badge")
-
-    dash = slot_public(auth, "/", "dashboard")
-    yeehaw_events = [e for e in dash.get("activity", []) if e["kind"] == "yeehaw"]
-    check(len(yeehaw_events) == 5
-          and all(int(e["task_id"]) == yeehaw_task for e in yeehaw_events),
-          "every yeehaw entry surfaces as its own timeline event kind")
-    check(yeehaw_events[0]["text"] == "🤖 yeehaw'd: Fifth hoedown",
-          "the newest yeehaw event leads the timeline rows")
-    check(all("second line" not in e["text"] for e in yeehaw_events),
-          "multiline yeehaw messages snippet to their first line")
-    referenced = [e for e in dash.get("activity", []) if int(e["task_id"]) == yeehaw_task]
-    check(len(referenced) > 0 and all(e.get("task_yeehaw") is True for e in referenced),
-          "dashboard timeline task references carry the badge flag")
-    plain_refs = [e for e in dash.get("activity", []) if int(e["task_id"]) == layout_task]
-    check(len(plain_refs) > 0 and all(e.get("task_yeehaw") is False for e in plain_refs),
-          "references to yeehaw-less tasks stay bare")
-    status, home_page = http("GET", "/", timeout=15)
-    check(status == 200 and 'stm-if="item.task_yeehaw"' in home_page
-          and "badge-yeehaw" in home_page,
-          "the dashboard renders the badge on timeline task references")
-
-    # Close the yeehaw fixture's round (a plain done push under its
-    # session): under the derived view (item 23 — busy iff an open
-    # session on an assigned/working task) the fixture round would
-    # hold the fake's saddle busy forever; the goal section below
-    # needs it free.
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": yeehaw_session, "state": "done",
-                            "summary": "yeehaw fixture round closed"})
-    check(wait_task_done(auth, yeehaw_task, timeout=30) is not None,
-          "the yeehaw fixture round closes (the derived busy view frees the fake)")
-
-    # (The old R10 scenario — branch_advanced riding /poll verdicts —
-    # died with the poll channel; the answers bundle pins the hint on
-    # its surviving surface, the resumed assignment payload.)
-
     print("== plan 12: the web surfaces — pile, timeline, queue, task page ==")
     # A goal parked in plan review feeds the unified dashboard pile;
     # goal events feed the timeline.
@@ -749,19 +637,15 @@ def run(ctx):
     task_page_html = http_get_page(auth, "/tasks/%d" % dash_run)
     check("Mandate" in task_page_html and 'stm-markdown="task.description"' in task_page_html,
           "the task page frames the description as the mandate")
-    check(re.search(r"'/terminal\?session_id=' \+ session\.id\"\s+target=\"_blank\"",
+    check(re.search(r"'/tasks/' \+ task\.id \+ '/log'\"\s+target=\"_blank\"",
                     task_page_html) is not None,
-          "the task page's session terminal links open in a new tab")
-    terminal_state = slot_public(auth, "/tasks/%d/terminal" % dash_run, "terminal")
-    session_rows = terminal_state["sessions"]
-    check(len(session_rows) >= 2
-          and all(("raw" in s) for s in session_rows),
-          "the terminal state lists every goal-run session with per-session raw flags")
-    check(int(terminal_state["session_id"]) == int(session_rows[0]["id"]),
-          "the terminal default selects the LATEST session (rows are newest-first)")
-    terminal_html = http_get_page(auth, "/tasks/%d/terminal" % dash_run)
-    check("requestedSessionId" in terminal_html and "data-session-id" in terminal_html,
-          "the terminal client owns the ?session_id= selection (query params never reach entrypoints)")
+          "the task page's event log links open in a new tab")
+    log_state = slot_public(auth, "/tasks/%d/log" % dash_run, "task_log")
+    check(log_state.get("exists") is True and int(log_state["task_id"]) == dash_run,
+          "the event log page state names the run task")
+    log_html = http_get_page(auth, "/tasks/%d/log" % dash_run)
+    check("EventSource" in log_html and "events/stream" in log_html,
+          "the log page subscribes to the task's SSE stream")
     run_detail = task_state(auth, dash_run)
     check(run_detail.get("is_goal_run") is True and run_detail.get("mode") == "orchestrate",
           "the goal agent's own run task identifies itself on its page")
@@ -831,7 +715,6 @@ def run(ctx):
           and queue_page.index("New task") < queue_page.index("<table"),
           "the New task expander sits above the tasks table")
     # Comments/blocks threads: newest first since the sweep. The
-    # yeehaw task above carries session comments; assert the detail's
     # comment list leads with the most recently added one.
     conn = db()
     conn.execute(

test/bundles/replay.py

@@ -4,13 +4,10 @@ from harness import *  # noqa: F401,F403 — the shared wire helpers
 
 
 # =============================================================================
-# Bundle: replay — plan 13 B1/B2: the daemon-global entry dedup and the
-# idempotent terminal/narrative projections. Replays the live-failure
-# post-mortem shapes directly against the push channels: an outbox
-# redelivery arriving under the WRONG session id after churn (and after
-# a web restart), and repeated failure pushes that must still project
-# exactly once. Includes this shard's one deliberate app restart (the
-# dedup must survive it — it is database state, not memory).
+# Bundle: replay — plan 13 B1/B2 shapes, rebuilt on the plan-17 ledger:
+# the idempotent events push (UNIQUE(session_id, seq) — database state,
+# so it survives a web restart), the session-churn redelivery shape, and
+# repeated failure pushes that must still project exactly once.
 # =============================================================================
 
 def run(ctx):
@@ -21,18 +18,11 @@ def run(ctx):
         conn = db()
         try:
             return conn.execute(
-                "select count(*) from session_commands where session_id = ?",
+                "select count(*) from events where session_id = ?",
                 (session,)).fetchone()[0]
         finally:
             conn.close()
 
-    def dedup_rows():
-        conn = db()
-        try:
-            return conn.execute("select count(*) from saddle_entry_dedup").fetchone()[0]
-        finally:
-            conn.close()
-
     def open_session_on_fake(task_id):
         """A fixture open session row on the shard's saddle (the churn
         shape: a fresh web-side run row the daemon never really drove)."""
@@ -48,7 +38,7 @@ def run(ctx):
         finally:
             conn.close()
 
-    print("== plan 13 B1: entry dedup is daemon-global, not per-session ==")
+    print("== plan 17 §7.2: event pushes are idempotent per session ==")
     # A first task runs to completion; its session is the replay source.
     first = create_queued(auth, {"project_id": "1", "title": "Replay source task",
                                  "priority": "2"})
@@ -59,34 +49,32 @@ def run(ctx):
         "select max(id) from sessions where task_id = ?", (first,)).fetchone()[0]
     conn.close()
 
-    # The outbox batch: two command entries the daemon already
-    # delivered once (their ids landed with the source session).
+    # The outbox batch: two events the daemon already delivered once.
     burst = [
-        {"id": 5001, "command": "make test", "output": "ok", "exit_code": 0,
-         "type": "command", "timestamp": "2026-09-11T00:00:00Z"},
-        {"id": 5002, "command": "grep -r flange src/", "output": "src/flange.vala",
-         "exit_code": 0, "type": "command", "timestamp": "2026-09-11T00:00:01Z"},
+        {"seq": 5001, "kind": "tool_call", "actor": "model",
+         "payload": {"turn": 7, "tool": "run_command",
+                     "args": {"command": "make test"}}},
+        {"seq": 5002, "kind": "tool_result", "actor": "harness",
+         "payload": {"turn": 7, "stdout": "ok", "exit_code": 0}},
     ]
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": src_session, "state": "done",
-                            "entries": burst})
-    check(status == 200 and rows_for(src_session) >= 2,
-          "the entries batch stores under its session")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": src_session, "events": burst})
+    check(status == 200 and int(body.get("stored") or 0) == 2,
+          "the event batch stores under its session")
     stored_once = rows_for(src_session)
-    dedup_after_first = dedup_rows()
 
     # Redelivery under the SAME session: consumed, nothing new.
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": src_session, "state": "done",
-                            "entries": burst})
-    check(status == 200 and rows_for(src_session) == stored_once
-          and dedup_rows() == dedup_after_first,
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": src_session, "events": burst})
+    check(status == 200 and int(body.get("stored") or 0) == 0
+          and rows_for(src_session) == stored_once,
           "a same-session redelivery stores nothing new (consumed silently)")
 
-    # Session churn: a brand-new session id on the same saddle replays
-    # the SAME entry ids — the post-mortem's mis-keyed outbox send.
-    # The entries' ids already populated another session, so the push
-    # may not create rows here either.
+    # Session churn: a mis-keyed outbox send under a brand-new session
+    # id replays the SAME seqs. The ledger's idempotency key is
+    # (session_id, seq) — the churned session is a different key, so
+    # the events store under it (the daemon contract makes the session
+    # id part of the key; the old daemon-global entry dedup is gone).
     conn = db()
     conn.execute(
         "insert into tasks (project_id, title, status, priority, created_at,"
@@ -97,26 +85,14 @@ def run(ctx):
         "select max(id) from tasks where title = 'Churn target task'").fetchone()[0]
     conn.close()
     churn_session = open_session_on_fake(churn_task)
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": churn_session, "state": "done",
-                            "entries": burst})
-    check(status == 200 and rows_for(churn_session) == 0,
-          "replay-under-wrong-session stores nothing new (daemon-global dedup)")
-    check(dedup_rows() == dedup_after_first,
-          "the consumed redelivery claims no fresh dedup rows")
-
-    # A fresh entry id under the churned session DOES store — the dedup
-    # keys on the entry id, never on the session.
-    fresh = [{"id": 5003, "command": "git status", "output": "clean",
-              "exit_code": 0, "type": "command", "timestamp": "2026-09-11T00:00:02Z"}]
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": churn_session, "state": "done",
-                            "entries": fresh})
-    check(status == 200 and rows_for(churn_session) == 1
-          and dedup_rows() == dedup_after_first + 1,
-          "a fresh entry id stores under any session of its saddle")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": churn_session, "events": burst})
+    check(status == 200 and int(body.get("stored") or 0) == 2
+          and rows_for(churn_session) == 2,
+          "a churned-session replay stores under the new key (per-session seq)")
 
-    # Web restart: the dedup survives (database state, not memory).
+    # Web restart: the idempotency survives (database constraint, not
+    # memory) — the same-session redelivery still stores nothing.
     app.terminate()
     app.wait(timeout=10)
     app = start_app()
@@ -127,17 +103,16 @@ def run(ctx):
     # re-establishes sleeping with a fresh ask (the real daemon's
     # idle-awake ask / 15-minute safety self-ask).
     fake.ask_now()
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": churn_session, "state": "done",
-                            "entries": burst + fresh})
-    check(status == 200 and rows_for(churn_session) == 1
-          and rows_for(src_session) == stored_once,
-          "a post-restart redelivery still stores nothing (dedup is persistent)")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": churn_session, "events": burst})
+    check(status == 200 and int(body.get("stored") or 0) == 0
+          and rows_for(churn_session) == 2,
+          "a post-restart redelivery still stores nothing (persistent)")
 
-    print("== plan 13 B2: duplicate failed entries project once (task) ==")
-    # The post-mortem's honest burst: the same failure entry redelivered
-    # under a churned session AND a second, DISTINCT failure entry id
-    # in the same session — one transition, one comment, ever.
+    print("== plan 13 B2: duplicate failed pushes project once (task) ==")
+    # The post-mortem's honest burst, rebuilt on the state channel: the
+    # failed state redelivered repeatedly — one transition, one
+    # summary, ever (terminal pushes are consumed silently).
     def agent_comments(task_id, body=None):
         conn = db()
         try:
@@ -162,40 +137,32 @@ def run(ctx):
         "select max(id) from tasks where title = 'Burst failure task'").fetchone()[0]
     conn.close()
     burst_session = open_session_on_fake(burst_task)
-    first_fail = [{"id": 6001, "command": "saddle failure",
-                   "output": "bash -n rejected the script", "exit_code": 1,
-                   "type": "failed", "timestamp": "2026-09-11T00:01:00Z"}]
     status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                            {"session_id": burst_session, "state": "failed",
-                            "entries": first_fail,
                             "summary": "bash -n rejected the script"})
     check(status == 200
           and task_state(auth, burst_task)["status"] == "failed",
-          "the failure entry + state pair fails the task")
+          "the failure state push fails the task")
     check(agent_comments(burst_task, "bash -n rejected the script") == 0
           and "bash -n rejected the script"
           in (task_state(auth, burst_task).get("summary") or ""),
           "the failure explanation lives ONCE — in tasks.summary, the terminal cache (S4: no duplicate agent comment)")
 
-    # The burst: a DISTINCT failure entry id in the same session, plus
-    # both ids redelivered under a churned session — nothing re-runs.
-    second_fail = [{"id": 6002, "command": "saddle failure",
-                    "output": "bash -n rejected the script", "exit_code": 1,
-                    "type": "failed", "timestamp": "2026-09-11T00:01:01Z"}]
+    # The burst: the same failure redelivered under the session AND a
+    # churned session — nothing re-runs.
     churned_burst = open_session_on_fake(burst_task)
     for target in (burst_session, churned_burst):
         status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                                {"session_id": target, "state": "failed",
-                                "entries": first_fail + second_fail,
                                 "summary": "bash -n rejected the script"})
         check(status == 200,
-              "the burst push under %s is consumed, not rejected" % (
+              "the failed-state push under %s is consumed, not rejected" % (
                   "its session" if target == burst_session else "a churned session"))
     check(task_state(auth, burst_task)["status"] == "failed"
           and agent_comments(burst_task) == 0
           and "bash -n rejected the script"
           in (task_state(auth, burst_task).get("summary") or ""),
-          "duplicate failed entries leave one transition, one summary, zero agent comments (S4)")
+          "duplicate failed pushes leave one transition, one summary, zero agent comments (S4)")
 
     print("== plan 13 B2: goal failure projects once ==")
     # Goal context: `saddle failure` fails the goal ONCE — the 115
@@ -229,27 +196,20 @@ def run(ctx):
         (run_task,)).fetchone()[0]
     conn.close()
 
-    goal_fail = [{"id": 7001, "command": "saddle failure",
-                  "output": "the harness collapsed", "exit_code": 1,
-                  "type": "failed", "timestamp": "2026-09-11T00:02:00Z"},
-                 {"id": 7002, "command": "saddle failure",
-                  "output": "the harness collapsed", "exit_code": 1,
-                  "type": "failed", "timestamp": "2026-09-11T00:02:01Z"}]
     status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                            {"session_id": run_session, "state": "failed",
-                            "entries": goal_fail, "summary": "the harness collapsed"})
+                            "summary": "the harness collapsed"})
     check(status == 200 and goal_state(burst_goal)["status"] == "failed",
           "the goal agent's failure fails the goal")
     check(goal_failed_events(burst_goal) == 1,
-          "a two-entry failure burst records exactly one goal-failed event")
+          "the failure push records exactly one goal-failed event")
 
-    # Redeliveries — same ids under the run's session AND under a
-    # churned session — plus the bare failed state again: still one.
+    # Redeliveries — under the run's session AND a churned session —
+    # plus the bare failed state again: still one.
     churned_run = open_session_on_fake(run_task)
     for target in (run_session, churned_run):
         status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
                                {"session_id": target, "state": "failed",
-                                "entries": goal_fail,
                                 "summary": "the harness collapsed"})
         check(status == 200, "the goal-failure redelivery under %s is consumed"
               % ("its session" if target == run_session else "a churned session"))
@@ -266,11 +226,11 @@ def run(ctx):
     check(burst_blocks == 0,
           "the goal-failed path raises ZERO blocks (nothing pins the pile)")
 
-    print("== plan 13 B4: the comment projection stores the body ==")
-    # The heredoc shape: the command is the pipe invocation, the OUTPUT
-    # is the body. The projections must carry the body — the command
-    # echo never becomes comment text, and the literal `saddle
-    # comment` string never renders outside the terminal.
+    print("== plan 17: agent remarks ride the ledger as comment events ==")
+    # The heredoc `cat << 'EOF' | saddle comment` shape: the daemon
+    # emits a comment event carrying the body. The body is the record —
+    # no command echo ever renders, and the goal page's current-activity
+    # line reads the newest remark.
     act(auth, slot_public(auth, "/goals", "goals")["create"], {
         "project_id": "1", "title": "Comment goal", "brief": "Reports cleanly.",
         "branch_name": "goal/comment"})
@@ -295,23 +255,24 @@ def run(ctx):
     conn.close()
 
     comment_body = "The flange weld holds. First sentence carries the information."
-    comment_entry = [{"id": 8001, "command": "cat << 'EOF' | saddle comment",
-                      "output": comment_body, "exit_code": 0, "type": "comment",
-                      "timestamp": "2026-09-11T00:03:00Z"}]
-    status, _ = saddle_api("POST", "/api/saddle/results", fake_token,
-                           {"session_id": comment_session, "state": "suspended",
-                            "entries": comment_entry})
-    check(status == 200, "the comment entry push is accepted")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": comment_session,
+                               "events": [{
+                                   "seq": 8001, "kind": "comment",
+                                   "actor": "model",
+                                   "payload": {"body": comment_body}}]})
+    check(status == 200 and int(body.get("stored") or 0) == 1,
+          "the comment event stores on the ledger")
     conn = db()
-    task_comment_bodies = [r[0] for r in conn.execute(
-        "select body from task_comments where task_id = ?", (comment_run,))]
     tables = {r[0] for r in conn.execute(
         "select name from sqlite_master where type = 'table'").fetchall()}
+    comment_events = conn.execute(
+        "select count(*) from events where scope_id = ? and kind = 'comment'",
+        (comment_run,)).fetchone()[0]
     conn.close()
-    check(task_comment_bodies == [comment_body],
-          "the run task's comment stores the body, not the command echo")
-    check("goal_comments" not in tables,
-          "the goal_comments table is gone (the run task's rows are the thread)")
+    check(comment_events == 1
+          and "goal_comments" not in tables,
+          "the ledger carries the remark (the goal_comments table is gone)")
     narrative_comments = [e["text"] for e in goal_state(comment_goal)["events"]
                           if e["kind"] == "comment"]
     check(len(narrative_comments) == 1
@@ -322,20 +283,16 @@ def run(ctx):
 
     state = goal_state(comment_goal)
     check(state["current_command"] == comment_body,
-          "the goal page's current-activity line shows the comment body")
-    check(state["comments"] and state["comments"][0]["body"] == comment_body
-          and any(e["kind"] == "comment"
-                  and e["text"].startswith("The flange weld holds.")
-                  for e in state["events"]),
-          "the goal slot carries the body in its comments and narrative")
+          "the goal page's current-activity line shows the remark body")
+    check(any(e["kind"] == "comment"
+              and e["text"].startswith("The flange weld holds.")
+              for e in state["events"]),
+          "the goal slot carries the remark in its narrative")
     page = http_get_page(auth, "/goals/%d" % comment_goal)
     check("saddle comment" not in page,
           "the rendered goal page never shows the command echo")
     check("event.kind == 'comment' ? ' footnote'" in page,
           "the narrative's comment events render with footnote styling")
-    check(page.index('stm-markdown="comment.body"')
-          < page.index('stm-text="comment.quote"'),
-          "the comment card shows the body before the quote")
 
     # Drain: nothing queued remains for the shard's shutdown.
     check(queued_count() == 0, "the replay shard drains its queue")

test/harness.py

@@ -249,14 +249,14 @@ def http(method, url, data=None, headers=None, timeout=30):
     # Retry dropped/refused connections: the freshly started app can
     # close a very early keep-alive connection once while warming up,
     # and its first seconds can wobble under load.
-    for attempt in range(3):
+    for attempt in range(6):
         try:
             with urllib.request.urlopen(request, timeout=timeout) as response:
                 return response.status, response.read().decode()
         except urllib.error.HTTPError as error:
             return error.code, error.read().decode()
         except (urllib.error.URLError, OSError):
-            if attempt >= 2:
+            if attempt >= 5:
                 raise
             time.sleep(0.75)
 
@@ -450,12 +450,30 @@ def command_rows(session):
     conn = db()
     try:
         return conn.execute(
-            "select command_text, type from session_commands"
+            "select kind, payload from events"
             " where session_id = ? order by id", (session,)).fetchall()
     finally:
         conn.close()
 
 
+def event_rows(task_id, kinds=None):
+    """The task's ledger rows (kind, payload JSON), oldest first."""
+    conn = db()
+    try:
+        if kinds:
+            marks = ",".join("?" for _ in kinds)
+            return conn.execute(
+                "select kind, payload from events"
+                " where scope_type = 'task' and scope_id = ? and kind in (%s)"
+                " order by id" % marks, (task_id,) + tuple(kinds)).fetchall()
+        return conn.execute(
+            "select kind, payload from events"
+            " where scope_type = 'task' and scope_id = ? order by id",
+            (task_id,)).fetchall()
+    finally:
+        conn.close()
+
+
 class FakeSaddle:
     """In-process saddle daemon implementing the wire contract.
 
@@ -540,45 +558,54 @@ class FakeSaddle:
             self.last_activity_epoch = int(time.time())
 
     def queue_output(self, entry_type, command, output):
-        """Pushes a typed entry as a round-results push (the daemon's
-        entry log riding the results channel)."""
-        with self.lock:
-            entry = {
-                "id": self.output_seq,
-                "command": command,
-                "output": output,
-                "exit_code": 1,
-                "type": entry_type,
-                "timestamp": "2026-09-06T00:00:00Z",
-            }
-            self.output_seq += 1
-        self.push_results(self.state if self.state in ("working",) else "working",
-                          entries=[entry])
+        """Pushes one command_output event home (the daemon's entry log
+        now rides the events channel; the old results-entries path is
+        gone)."""
+        self.push_events([{"kind": "command_output",
+                           "payload": {"command": command, "text": output}}])
 
     # --- plan 10 phase 2: the push channels ---
 
-    def push_chunks(self, chunks):
-        """Pushes raw PTY chunks home: elements are texts (assigned
-        fresh seqs) or {seq, text} dicts (a redelivery or explicit
-        corpus). Pushes are suppressed while silent/hold_state (the
-        staleness tests' silent runner)."""
+    def push_events(self, events):
+        """Pushes a batch of ledger events home (plan 17 §7.2): each
+        element a {seq?, kind, actor?, ts?, payload?} dict; a missing
+        seq is assigned from the per-session monotonic counter.
+        Pushes are suppressed while silent/hold_state (the staleness
+        tests' silent runner)."""
         if self.dead or self.silent or self.token is None or self.session_id is None:
             return
         batch = []
         with self.lock:
-            for chunk in chunks:
-                if isinstance(chunk, dict):
-                    batch.append({"seq": int(chunk["seq"]), "text": chunk["text"]})
-                else:
-                    batch.append({"seq": self.next_push_seq(), "text": chunk})
+            for event in events:
+                item = dict(event)
+                item["seq"] = int(item.pop("seq", 0) or self.next_push_seq())
+                item.setdefault("kind", "progress")
+                item.setdefault("actor", "model")
+                item.setdefault("payload", {})
+                batch.append(item)
         try:
             saddle_api("POST", "/api/saddle/chunks", self.token,
-                       {"session_id": self.session_id, "chunks": batch})
+                       {"task_id": self.running_task,
+                        "session_id": self.session_id, "events": batch})
         except (OSError, ValueError):
             pass  # app restarting / shutting down mid-test
 
+    def push_chunks(self, chunks):
+        """Legacy-shaped helper: raw texts (or {seq, text} dicts, a
+        redelivery) push as progress events carrying the text."""
+        batch = []
+        for chunk in chunks:
+            if isinstance(chunk, dict):
+                batch.append({"seq": chunk.get("seq"),
+                              "kind": chunk.get("kind", "progress"),
+                              "payload": chunk.get("payload",
+                                                   {"text": chunk.get("text", "")})})
+            else:
+                batch.append({"kind": "progress", "payload": {"text": chunk}})
+        self.push_events(batch)
+
     def next_push_seq(self):
-        # The daemon-side chunk counter for ad-hoc pushes.
+        # The daemon-side per-session event counter for ad-hoc pushes.
         self._push_seq = getattr(self, "_push_seq", 0) + 1
         return self._push_seq
 
@@ -594,8 +621,6 @@ class FakeSaddle:
         if target is None:
             return None
         payload = {"session_id": target, "state": state}
-        if entries is not None:
-            payload["entries"] = entries
         if block is not None:
             payload["block"] = block
             if state == "blocked":
@@ -711,14 +736,30 @@ class FakeSaddle:
             self.goal_behaviour = self.goal_script.pop(0) if self.goal_script else ("suspend",)
         self.set_state("failing" if self.fail_next else "working")
         self.fail_next = False
-        # The round's opening telemetry: the flushed-so-far raw PTY
-        # batch plus the working state push (>= 2 flushes per round
-        # and the 10s timer are the real daemon's contract; the fake
-        # pushes its batch up front and heartbeats while working).
+        # The round's opening telemetry: the §7.2 event batch (turn 1:
+        # lifecycle, turn text, tool call + result, usage) plus the
+        # working state push. A new session restarts the per-session
+        # seq counter; a scripted corpus supplements the batch.
+        self._push_seq = 0
+        opening = [
+            {"kind": "lifecycle",
+             "actor": "harness",
+             "payload": {"action": "start", "text": "round started"}},
+            {"kind": "turn_text",
+             "payload": {"turn": 1, "text": "Starting the assigned task."}},
+            {"kind": "tool_call",
+             "payload": {"turn": 1, "tool": "run_command",
+                         "args": {"command": "echo hello"}}},
+            {"kind": "tool_result",
+             "actor": "harness",
+             "payload": {"turn": 1, "stdout": "hello\n", "exit_code": 0}},
+            {"kind": "usage",
+             "actor": "harness",
+             "payload": {"turn": 1, "input_tokens": 120, "output_tokens": 30}},
+        ]
         if self.scripted_corpus is not None:
-            self.push_chunks([dict(chunk) for chunk in self.scripted_corpus])
-        else:
-            self.push_chunks([self.PROMPT, "echo hello\r\n", "hello\r\n"])
+            opening = [dict(chunk) for chunk in self.scripted_corpus]
+        self.push_events(opening)
         self.push_results("working")
 
     class Handler(BaseHTTPRequestHandler):