Browse Source

protocol-native integration pass: goal-era arc end to end against the real daemon — the harness speaks tool calls (mock script per §7 grammar: narrated probes, question via run_command's saddle block, submit_plan/create_task/enqueue+suspend, the worker's five identical probes tripping the loop warning, finish, the coalesced merge/push/submit_report resume) and asserts the ledger world (turn_text/tool_call/tool_result/usage events, §7.2 kind set, seq exactly-once on chunks, the event log page + SSE replay, tool results as role=tool turns, the retired results-entries array consumed silently); web fixes surfaced by the pass: a review-kind block push on a goal run routes to submit_completion (the daemon's submit_report done-park now reaches completion_review with its goal-scoped narrative row), the narrative derives block-raised from the run task's question/RFC raises, the U2 mechanical rows (session context restored, parked-until) persist as system comment events on resume/park with the goal thread filtered like the task page, and the replay bundle's remark query pins the model-actor row; integration 125/0, 13 bundles sequential 966/0

clanker 1 tháng trước cách đây
mục cha
commit
7007d4b94c

+ 5 - 0
src/Services/AllocationService.vala

@@ -403,6 +403,11 @@ namespace ClankerSaddle.Services {
                     // and ends the parking window.
                     resume_notice = yield goals.take_resume_notice(task.goal_id);
                     yield tasks.clear_suspension(task.id);
+                    // Plan 16 S5 (U2): the restore footnote persists in
+                    // the audit ledger (a system comment event) and
+                    // never renders on a user-facing timeline.
+                    yield tasks.add_system_comment(task.id,
+                        "session context restored — task resumed");
                 } else if (claim.suspended_resume) {
                     // Goal 4 task 1 (Force Wake): a force-woke PLAIN
                     // task's resume carries the explicit human-wake

+ 15 - 1
src/Services/GoalService.vala

@@ -908,7 +908,7 @@ namespace ClankerSaddle.Services {
             var run = yield goal_run_task(goal_id);
             if (run != null) {
                 foreach (var record in yield events.list_for_task(((!)run).id,
-                        { EVENT_VERDICT, EVENT_ANSWER, EVENT_COMMENT })) {
+                        { EVENT_VERDICT, EVENT_ANSWER, EVENT_COMMENT, EVENT_STATE_CHANGE })) {
                     if (record.created_at == null) {
                         continue;
                     }
@@ -922,6 +922,20 @@ namespace ClankerSaddle.Services {
                         case EVENT_ANSWER:
                             kind = "block-answered";
                             break;
+                        case EVENT_STATE_CHANGE:
+                            // The run task's question/RFC raises surface
+                            // as narrative rows (the derived equivalent
+                            // of the old goal_events raise entries; plan
+                            // and completion raises carry their own
+                            // goal-scoped events and stay out here).
+                            string raised = EntryJson.string_member(payload, "block_type");
+                            if (raised != ClankerSaddle.Services.BLOCK_QUESTION
+                                    && raised != ClankerSaddle.Services.BLOCK_RFC) {
+                                continue;
+                            }
+                            kind = "block-raised";
+                            text = EntryJson.string_member(payload, "question");
+                            break;
                         default:
                             string author = EntryJson.string_member(payload, "author");
                             text = EntryJson.string_member(payload, "body");

+ 9 - 0
src/Services/SaddleService.vala

@@ -1369,6 +1369,15 @@ namespace ClankerSaddle.Services {
                         if (block_string_or((!)block, "type", "block_type") == BLOCK_PLAN) {
                             yield goal_service.submit_plan(goal_id,
                                 EntryJson.optional_string_member((!)block, "body") ?? question);
+                        } else if (block_string_or((!)block, "type", "block_type") == BLOCK_REVIEW) {
+                            // The daemon's submit_report done-park: a
+                            // review-kind block on the goal run IS the
+                            // completion submission (submit_completion
+                            // adopts the raised row — one row, whichever
+                            // path landed first — and drives the
+                            // completion_review transition).
+                            yield goal_service.submit_completion(goal_id,
+                                EntryJson.optional_string_member((!)block, "body") ?? question);
                         } else {
                             yield goal_service.raise_agent_block(goal_id, question,
                                 block_string_or((!)block, "type", "block_type"),

+ 5 - 0
src/Services/TaskService.vala

@@ -1196,6 +1196,11 @@ namespace ClankerSaddle.Services {
             yield pulse.bump(task_id, "task suspended");
             yield append_state_change(task_id, STATUS_WORKING, STATUS_SUSPENDED,
                 "Parked — awaiting the resume", EVENT_ACTOR_SYSTEM);
+            // Plan 16 S5 (U2): the mechanical park notice persists in
+            // the audit ledger (a system comment event) and never
+            // renders on a user-facing timeline.
+            yield add_system_comment(task_id,
+                "goal agent parked until an owned task completes or fails");
         }
 
         /**

+ 7 - 1
src/entrypoints/GoalDetailEntrypoint.vala

@@ -771,8 +771,14 @@ namespace ClankerSaddle {
             // Plan 15 C2: the goal's comment thread IS the run
             // task's task_comments (injection rides the existing
             // delivery paths) — the agent's pushes and the human's
-            // composer land on the same rows.
+            // composer land on the same rows. Plan 16 S5 (U2):
+            // mechanical rows (restores, park notices) never reach
+            // the user-facing thread — the audit ledger keeps them.
             foreach (var row in yield ctx.goals.list_comments(goal.id)) {
+                if (row.author == COMMENT_AUTHOR_SYSTEM
+                        && comment_is_mechanical(row.body)) {
+                    continue;
+                }
                 view.comments.add(new GoalCommentView() {
                     author = row.author,
                     body = row.body,

+ 4 - 2
test/bundles/replay.py

@@ -267,12 +267,14 @@ def run(ctx):
     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'",
+        "select count(*) from events where scope_id = ? and kind = 'comment'"
+        " and actor = 'model'",
         (comment_run,)).fetchone()[0]
     conn.close()
     check(comment_events == 1
           and "goal_comments" not in tables,
-          "the ledger carries the remark (the goal_comments table is gone)")
+          "the ledger carries the remark (the goal_comments table is gone;"
+          " system bookkeeping comments are separate)")
     narrative_comments = [e["text"] for e in goal_state(comment_goal)["events"]
                           if e["kind"] == "comment"]
     check(len(narrative_comments) == 1

+ 289 - 225
test/integration_test.py

@@ -5,42 +5,42 @@ Drives the web app the way statum.js does (entrypoint state loads plus
 action POSTs carrying the auth slot and sealed private blobs) while a
 REAL saddle daemon runs a REAL goal against a real git checkout under
 WORKSPACE, powered by saddle/test/mock_llm.py with a deterministic
-script — both binaries, the true cross-binary contract proof (the
-per-repo fake-peer suits live in web_stack_test.py / saddle e2e).
+tool-call script — both binaries, the true cross-binary contract proof
+(the per-repo fake-peer suits live in web_stack_test.py / saddle e2e).
 
-The consolidated goal-era arc (plan 15, re-baselined for C+D):
+The protocol-native goal-era arc (PLAN-17; every agent turn is a tool
+call, questions ride `saddle block` through run_command):
 1. goal create -> start: the goal-run assignment reaches the daemon
    over the ask/allocate channel with the full plan-12 payload (kind/
    mode/creator/goal context/ephemeral branches for workers)
-2. a commentary-only round and a trailing-comment round feed VERBATIM
-   (comment-policy-free inputs, plan 15 D1) before the pre-plan
-   question block -> answered on the goal page -> the ONE answer
-   resolves the goal-run task's single task_blocks row and the parked
-   goal agent resumes with the verdict wording
-3. plan submit -> reject-with-feedback -> revision -> approval (the
+2. a narrated probe round (turn_text + tool_call/tool_result events in
+   the ledger) precedes the pre-plan question block -> answered on the
+   goal page -> the ONE answer resolves the goal-run task's single
+   task_blocks row and the parked goal agent resumes with the verdict
+3. submit_plan -> reject-with-feedback -> revision -> approval (the
    agreed verdict words ride the resume)
-4. task fan-out through the real /api/saddle/task/create + enqueue
-   callbacks; the worker's DEGENERATE first response is
-   validator-caught (plan 15 D2 — never executed, one receipt, the
-   session continues); the clean resend runs in its ephemeral
-   worktree branch cut from the goal clone, commits, completes —
-   worktree auto-pruned, branch kept
-5. the terminal transition (the goals row's last_terminal_watermark,
-   plan 15 C3 — goal_events is gone) coalesces (~SADDLE_GOAL_RESUME_S)
-   into the goal agent's batched resume; it merges, deletes the task
-   branch, pushes the goal branch (push-before-review) and submits
-   the completion report
-6. completion review accept — the live per-repo push flags gate it
-   (plan 15 C4): the accept REFUSES while origin is behind, passes
-   once the branch is re-pushed and the probe re-cached -> goal
-   complete, run task done
-7. the goal agent's terminal sessions (SSE replay), the audit trail
-   (the DERIVED goal narrative, typed session_commands, the
-   harness-authored bookkeeping riding the RAW terminal only), the
-   /status goal_branches cache sampled by the manual saddle refresh
-8. the push wire surface (401s on the id-less routes, typed entries
-   and block raises riding POST /api/saddle/results with per-entry
-   exactly-once dedup, the deleted poll-era endpoints 404ing) and the
+4. task fan-out through the create_task/enqueue_task tools (the real
+   /api/saddle/task/create + enqueue callbacks); the worker's repeated
+   identical run_command probes trip the loop warning (never fatal),
+   then the clean run works in its ephemeral worktree branch cut from
+   the goal clone, commits, finishes — worktree auto-pruned, branch
+   kept
+5. the terminal transition (the goals row's last_terminal_watermark)
+   coalesces (~SADDLE_GOAL_RESUME_S) into the goal agent's batched
+   resume; it merges, deletes the task branch, pushes the goal branch
+   (push-before-review) and submit_report enters completion review
+6. completion review accept — the live per-repo push flags gate it:
+   the accept REFUSES while origin is behind, passes once the branch
+   is re-pushed and the probe re-cached -> goal complete, run task done
+7. the audit trail: the DERIVED goal narrative, the unified events
+   ledger (turn cards with tool_call/tool_result, usage rows, the §7.2
+   kind set, web-originated-only comment kinds), the event log page +
+   SSE stream, the /status goal_branches cache sampled by the manual
+   saddle refresh
+8. the push wire surface (401s on the id-less routes, the §7.2 chunks
+   ingestion with (session_id, seq) exactly-once, block raises riding
+   POST /api/saddle/results, the retired results-entries array
+   consumed silently, the deleted poll-era endpoints 404ing) and the
    stm-markdown renderer
 
 Prerequisites (see README.md "Integration testing"): the web app
@@ -242,11 +242,12 @@ def db_query(sql, params=()):
         db.close()
 
 
-def capture_user_turns():
-    """Every user turn the mock received, oldest first (the mock's
-    --capture file: one JSON line per request). The D-side receipts
-    (round outputs) ride these turns back to the model."""
-    turns = []
+def capture_messages():
+    """Every message the mock received, oldest first (the mock's
+    --capture file: one JSON line per request). Tool results ride back
+    to the model as role=tool turns; harness notices (loop warnings,
+    verdicts) as role=user turns."""
+    messages = []
     if os.path.exists(CAPTURE_PATH):
         with open(CAPTURE_PATH) as handle:
             for line in handle:
@@ -255,9 +256,10 @@ def capture_user_turns():
                 except ValueError:
                     continue
                 for message in arrival.get("messages", []):
-                    if message.get("role") == "user":
-                        turns.append(message.get("content") or "")
-    return turns
+                    if message.get("role") in ("user", "tool"):
+                        messages.append((message.get("role"),
+                                         message.get("content") or ""))
+    return messages
 
 
 def prepare_workspace():
@@ -284,74 +286,91 @@ def prepare_workspace():
     subprocess.run(["git", "-C", DEMO, "push", "origin", "main"], check=True, capture_output=True)
 
 
+def tool_call(name, arguments):
+    return "tool:%s:%s" % (name, json.dumps(arguments))
+
+
+def parse_call(spec):
+    """One mock call spec back into the {name, arguments} shape the
+    JSON array and content+calls grammars expect."""
+    name, arguments = spec[len("tool:"):].split(":", 1)
+    return {"name": name, "arguments": json.loads(arguments)}
+
+
 def mock_argv():
-    heredoc = "cat << 'SADDLE_EOF' | %s\n%%s\nSADDLE_EOF"
-    fanout = (
-        "# Fanning the plan out\n"
-        "out=$(cat << 'SADDLE_EOF' | saddle task create --title 'Write the greeting' --mode implement\n"
-        "Create greeting.txt containing hello-from-the-goal.\nSADDLE_EOF)\n"
-        "tid=$(echo \"$out\" | grep -oE '[0-9]+' | head -1)\n"
-        "saddle task enqueue \"$tid\"\n"
-        "saddle suspend"
-    )
+    """The protocol-native arc, one scripted assistant turn each,
+    consumed in strict arrival order (single daemon, sequential
+    sessions: goal run parks before the worker starts, the worker
+    finishes before the goal agent's terminal-event resume)."""
+    question = ("cat << 'SADDLE_EOF' | saddle block question \"Demo only\" \"Whole workspace\"\n"
+                "Which scope should the greeting cover? The demo repo is small; confirm the blast radius.\n"
+                "SADDLE_EOF")
     responses = [
-        # R1 (fresh goal agent): commentary-ONLY (plan 15 D1) — fed
-        # verbatim, receipt [0 command lines] plus the factual
-        # no-commands line, conversation continues
-        "# Reading the brief before touching anything\n"
-        "# the demo repo is tiny; the greeting belongs in greeting.txt\n"
-        "# nothing worth running yet — probing comes next",
-        # R2: investigate with TRAILING comments after the commands
-        # (comments-after-commands feeds exactly like leading ones),
-        # then block on a scope question
-        "# Investigating before planning\nls -la\ngit log --oneline | head -3\n"
-        "# two commits and one file — the blast radius is greeting.txt alone\n"
-        + (heredoc % 'saddle block question "Demo only" "Whole workspace"') % (
-            "Which scope should the greeting cover? The demo repo is small; confirm the blast radius."),
-        # W1 (the worker's first round): a DEGENERATE repetition (plan
-        # 15 D2) — the validator catches it before feeding; nothing
-        # executes, one rejection receipt, the round is consumed
-        "# Starting the mandate\n" + "echo probing-the-workspace\n" * 12,
-        # W2 (the clean resend): verify the ephemeral branch, commit, done
-        "# Working the mandate\ngit branch --show-current\n"
-        "echo hello-from-the-goal > greeting.txt\ngit add -A\n"
-        "git commit -m 'greeting from the goal'\n"
-        + (heredoc % "saddle done") % (
-            "# Completion report\n\n- changed: greeting.txt\n"
-            "- verified: committed on the ephemeral branch"),
+        # G1 (fresh goal agent): narrated probe — turn_text plus one
+        # tool call, the protocol-native shape of the old commentary
+        # round
+        json.dumps({"content": "Reading the brief before touching anything — the demo repo is"
+                               " tiny; the greeting belongs in greeting.txt.",
+                    "calls": [parse_call(tool_call("run_command", {"command": "ls -la"}))]}),
+        # G2: investigate, then block on a scope question (the saddle
+        # CLI stays the question surface; run_command feeds the PTY)
+        json.dumps({"content": "Two commits and one file — the blast radius is greeting.txt"
+                               " alone; confirming the scope with the human.",
+                    "calls": [parse_call(tool_call("run_command", {"command": "git log --oneline | head -3"}))]}),
+        tool_call("run_command", {"command": question}),
+        # G3 (answer resume): submit the plan -> plan review park
+        tool_call("submit_plan", {"markdown": "# Plan v1\n\n1. Add greeting.txt to the demo repo."}),
+        # G4 (feedback resume): the revision incorporating the feedback
+        tool_call("submit_plan", {"markdown": "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n"
+                                              "2. Export it for reuse."}),
+        # G5/G6 (approval resume): fan out through the real callbacks,
+        # then park suspended (enqueue + suspend tolerate sharing one
+        # message — the harness executes them in order)
+        tool_call("create_task", {"title": "Write the greeting", "mode": "implement",
+                                  "brief": "Create greeting.txt containing hello-from-the-goal."}),
+        json.dumps([parse_call(tool_call("enqueue_task", {"task_id": 2})),
+                    parse_call(tool_call("suspend", {}))]),
+        # W1..W5 (the worker): five IDENTICAL probes — the loop
+        # detector injects its warning user turn after the fifth,
+        # never refusing execution
+    ] + [tool_call("run_command", {"command": "echo probing-the-workspace"})] * 5 + [
+        # W6: post a durable remark (§7.2 comment kind — the body is
+        # the record; the web renders it as a narrative footnote)
+        tool_call("add_comment", {"body": "Probing done — the mandate starts on the"
+                                          " ephemeral branch."}),
+        # W7: the clean mandate — verify the ephemeral branch, commit
+        json.dumps({"content": "Loop flagged — switching to the mandate on the ephemeral branch.",
+                    "calls": [parse_call(tool_call("run_command", {"command": "git branch --show-current"}))]}),
+        tool_call("run_command", {"command": "echo hello-from-the-goal > greeting.txt && git add -A"
+                                             " && git commit -m 'greeting from the goal'"}),
+        tool_call("finish", {"status": "done",
+                             "summary": "# Report\n\n- changed: greeting.txt\n"
+                                        "- verified: committed on the ephemeral branch"}),
+        # G7 (the coalesced terminal-event resume): hold the working
+        # window open a moment (the /status branch probe samples it),
+        # then merge, prune, push
+        tool_call("run_command", {"command": "sleep 2"}),
+        tool_call("run_command", {"command": "git merge --no-edit $(git for-each-ref"
+                                             " --format='%(refname:short)' refs/heads/task/*)"
+                                             " && git branch -D $(git for-each-ref"
+                                             " --format='%(refname:short)' refs/heads/task/*)"
+                                             " && git push -u origin $(git branch --show-current)"}),
+        # G8: submit_report -> completion review park
+        tool_call("submit_report", {"markdown": "# Goal completion report\n\n"
+                                                "- achieved: greeting.txt added via one worker task\n"
+                                                "- verified: the goal branch is pushed to origin"}),
     ]
     argv = []
     for response in responses:
         argv += ["-r", response]
     argv += [
-        # the question's verdict ("Human selected: Demo only") -> plan v1
-        "--on-answer", (heredoc % "saddle plan") % (
-            "# Plan v1\n\n1. Add greeting.txt to the demo repo."),
-        # the plan-feedback verdict -> the revision (the export step
-        # incorporates the harness's feedback)
-        "--on-plan-feedback", (heredoc % "saddle plan") % (
-            "# Plan v2\n\n1. Add greeting.txt to the demo repo.\n2. Export it for reuse."),
-        # the approval verdict -> task fan-out + suspend
-        "--on-plan-approved", fanout,
-        # the batched "Goal update" resume -> merge, prune, push, complete
-        "--answer-contains", "Goal update",
-        "--on-answer-seen",
-        "# Merging the finished task, then completing (the sleep holds the\n"
-        "# working state open so /status can sample the branch state)\n"
-        "sleep 4\n"
-        "git merge --no-edit $(git for-each-ref --format='%(refname:short)' refs/heads/task/*)\n"
-        "git branch -D $(git for-each-ref --format='%(refname:short)' refs/heads/task/*)\n"
-        "git push -u origin $(git branch --show-current)\n"
-        + (heredoc % "saddle done") % (
-            "# Goal completion report\n\n- achieved: greeting.txt added via one worker task\n"
-            "- verified: the goal branch is pushed to origin"),
         "--on-query", "The hostname is %s." % HOSTNAME,
         # Every reply lingers half a second: the local mock answers so
         # instantly that a worker's whole run (its DERIVED-busy open
         # window) can close inside one sampling interval otherwise.
         "--delay", "0.5",
         # receipt forensics: every prompt the mock received, so the
-        # harness can assert the D-side receipts reached the model
+        # harness can assert the tool results reached the model
         "--capture", CAPTURE_PATH,
     ]
     return argv
@@ -582,13 +601,20 @@ def main():
               "the saddle card derives busy from the worker's open session (no cached row)")
 
         print("== the worker's ephemeral worktree ==")
+        # The branch probe rides the ledger: the tool_call event carries
+        # the git command, its tool_result event the branch on stdout.
         branch_rows = wait_for(lambda: db_query(
-            "select sc.output_text from session_commands sc join sessions s"
-            " on s.id = sc.session_id where s.task_id = ?"
-            " and sc.command_text like '%git branch --show-current%'", (worker_task,)) or False,
+            "select payload from events where scope_type = 'task' and scope_id = ?"
+            " and kind = 'tool_result' and payload like ?",
+            (worker_task, "%" + ephemeral + "%")) or False,
             timeout=90, what="the worker's branch probe")
         check(branch_rows and ephemeral in (branch_rows[0][0] or ""),
-              "the worker runs inside its ephemeral branch %s" % ephemeral)
+              "the worker runs inside its ephemeral branch %s (tool_result stdout)" % ephemeral)
+        branch_call = db_query(
+            "select count(*) from events where scope_type = 'task' and scope_id = ?"
+            " and kind = 'tool_call' and payload like '%git branch --show-current%'",
+            (worker_task,))[0][0]
+        check(branch_call == 1, "the branch probe's tool_call event sits in the ledger")
         worktree_dir = os.path.join(WORKSPACE, "task-%d" % worker_task)
         check(wait_for(lambda: task_state(auth, worker_task)["status"] == "done" or False,
                        timeout=120, what="worker done") is not None,
@@ -695,31 +721,60 @@ def main():
         check(final.get("completion_report") == winning_body,
               "the accept canonises the winning block's body as goals.completion_report (S1)")
 
-        print("== audit trail (the derived narrative) ==")
+        print("== audit trail (the derived narrative + the unified ledger) ==")
         narrative = goal_state(auth, goal1)["events"]
         kinds = [e["kind"] for e in narrative]
         for expected in ("started", "block-raised", "plan-submitted", "plan-rejected",
-                         "plan-approved", "task-created", "task-done",
-                         "completion-submitted", "goal-approved"):
+                          "plan-approved", "task-created", "task-done",
+                          "completion-submitted", "goal-approved"):
             check(expected in kinds, "the derived narrative records %s" % expected)
         check("system" not in kinds,
-              "mechanical system rows never reach the user-facing narrative (U2: raw log only)")
+              "mechanical system rows never reach the user-facing narrative (U2)")
         check(all(k != "task-enqueued" for k in kinds),
               "timestamp-less bookkeeping never returns (task-enqueued is gone)")
         check(db_query(
             "select count(*) from sqlite_master where type = 'table'"
             " and name = 'goal_events'")[0][0] == 0,
             "the goal_events table is gone (nothing is written on ingest)")
-        run_types = {row[0] for row in db_query(
-            "select distinct sc.type from session_commands sc join sessions s"
-            " on s.id = sc.session_id where s.task_id = ?", (run_task,))}
-        check({"bash", "block", "plan"} <= run_types,
-              "the goal agent's terminal carries typed entries (%s)" % sorted(run_types))
-        worker_types = {row[0] for row in db_query(
-            "select distinct sc.type from session_commands sc join sessions s"
-            " on s.id = sc.session_id where s.task_id = ?", (worker_task,))}
-        check({"bash", "done"} <= worker_types,
-              "the worker's terminal carries typed entries (%s)" % sorted(worker_types))
+        check(db_query(
+            "select count(*) from sqlite_master where type = 'table'"
+            " and name = 'session_commands'")[0][0] == 0,
+            "the session_commands table is gone (the ledger is the record)")
+        run_kinds = {row[0] for row in db_query(
+            "select distinct kind from events where scope_type = 'task' and scope_id = ?",
+            (run_task,))}
+        check({"turn_text", "tool_call", "tool_result", "usage", "lifecycle"} <= run_kinds,
+              "the goal agent's ledger carries the protocol kinds (%s)" % sorted(run_kinds))
+        worker_kinds = {row[0] for row in db_query(
+            "select distinct kind from events where scope_type = 'task' and scope_id = ?",
+            (worker_task,))}
+        check({"turn_text", "tool_call", "tool_result", "usage"} <= worker_kinds,
+              "the worker's ledger carries the protocol kinds (%s)" % sorted(worker_kinds))
+        daemon_kinds = {row[0] for row in db_query(
+            "select distinct kind from events where session_id is not null")}
+        check(daemon_kinds <= {"turn_text", "tool_call", "tool_result", "progress",
+                               "comment", "lifecycle", "usage", "state_change"},
+              "every daemon-pushed kind is a §7.2 kind (%s)" % sorted(daemon_kinds))
+        remark_rows = db_query(
+            "select actor, payload from events where kind = 'comment'"
+            " and session_id is not null and scope_id = ?", (worker_task,))
+        check(len(remark_rows) == 1 and remark_rows[0][0] == "model"
+              and "the mandate starts on the ephemeral branch" in (remark_rows[0][1] or ""),
+              "the add_comment remark rides the ledger as a model-actor comment event")
+        narrative_remarks = [e["text"] for e in goal_state(auth, goal1)["events"]
+                             if e["kind"] == "comment"]
+        check(any(text.startswith("Probing done") for text in narrative_remarks),
+              "the goal narrative renders the worker's remark as a comment row")
+        plan_call = db_query(
+            "select count(*) from events where scope_type = 'task' and scope_id = ?"
+            " and kind = 'tool_call' and payload like '%submit_plan%'", (run_task,))[0][0]
+        check(plan_call >= 2, "both plan submissions sit as tool_call events")
+        usage_rows = db_query(
+            "select payload from events where kind = 'usage' and scope_type = 'task'"
+            " and scope_id = ? limit 1", (worker_task,))
+        check(bool(usage_rows) and "prompt_tokens" in (usage_rows[0][0] or "")
+              and "elapsed_ms" in (usage_rows[0][0] or ""),
+              "usage events carry prompt/completion tokens and elapsed time")
 
         print("== the terminal watermark (C3) drives the resume ==")
         # The fresh-DB schema diff (the joint-QA gate): the removed
@@ -745,18 +800,17 @@ def main():
               "the goals row's terminal watermark advanced past the worker's transition"
               " (the resume consumed it)")
 
-        print("== the harness bookkeeping rides the RAW terminal (C2 + U2) ==")
-        # Plan 16 S5 (U2): the mechanical rows (restores, park
-        # notices) author NO comment rows — they persist as typed
-        # 'system' session_commands (the raw terminal log) and never
+        print("== the harness bookkeeping stays ledger-internal (C2 + U2) ==")
+        # Plan 16 S5 (U2): the mechanical rows (restores, park notices)
+        # author NO user-facing comment rows — they persist as
+        # system-authored comment events (the unified ledger) and never
         # render on a user-facing timeline.
-        raw_system = db_query(
-            "select sc.command_text, sc.output_text from session_commands sc"
-            " join sessions s on s.id = sc.session_id where s.task_id = ?"
-            " and sc.type = 'system'", (run_task,))
-        check(any("session context restored" in (row[1] or "") for row in raw_system)
-              and any("parked until an owned task" in (row[1] or "") for row in raw_system),
-              "the harness bookkeeping (restore, park) persists in the raw terminal log")
+        mechanical = db_query(
+            "select count(*) from events where kind = 'comment' and actor = 'system'"
+            " and (payload like '%session context restored%'"
+            " or payload like '%parked until%')")[0][0]
+        check(mechanical >= 1,
+              "the harness bookkeeping (restore, park) persists as system comment events")
         thread = goal_state(auth, goal1)["comments"]
         check(not any("session context restored" in c["body"] or "parked until" in c["body"]
                       for c in thread),
@@ -778,77 +832,72 @@ def main():
               "every run/worker session closed exactly once — parks stamp outcome 'parked',"
               " none dangle open (%s)" % sorted({row[0] for row in outcomes}))
 
-        print("== D-side through both binaries: commentary + degenerate ==")
-        run_bash = db_query(
-            "select sc.command_text from session_commands sc join sessions s"
-            " on s.id = sc.session_id where s.task_id = ? and sc.type = 'bash'",
-            (run_task,))
-        check(any("nothing worth running yet" in (row[0] or "")
-                  and "Reading the brief" in (row[0] or "") for row in run_bash),
-              "the narration-only response was fed verbatim (comments are style, not policy)")
-        check(any("git log --oneline" in (row[0] or "")
-                  and "blast radius is greeting.txt alone" in (row[0] or "")
-                  for row in run_bash),
-              "trailing comments after commands rode the same round (no rejection)")
-        rejected = db_query(
-            "select sc.command_text, sc.output_text from session_commands sc"
-            " join sessions s on s.id = sc.session_id"
-            " where sc.type = 'rejected' and s.task_id in (?, ?)",
-            (run_task, worker_task))
-        check(len(rejected) == 1 and rejected[0][0] == "degenerate response"
-              and "degenerate signature" in (rejected[0][1] or "")
-              and "echo probing-the-workspace" in (rejected[0][1] or "")
-              and "Resend the intended action only" in (rejected[0][1] or ""),
-              "exactly ONE rejected entry — the degenerate repetition, caught with its receipt")
-        worker_bash = db_query(
-            "select sc.command_text from session_commands sc join sessions s"
-            " on s.id = sc.session_id where s.task_id = ? and sc.type = 'bash'",
-            (worker_task,))
-        check(not any("probing-the-workspace" in (row[0] or "") for row in worker_bash),
-              "the degenerate script never executed (validator-caught before feeding)")
-        turns = capture_user_turns()
-        check(any(turn.startswith("[0 command lines executed")
-                  and "no commands ran — narrate in the same response as your next command" in turn
-                  for turn in turns),
-              "the narration round's receipt reached the model (no-commands line)")
-        check(any("degenerate signature" in turn and "Resend the intended action only" in turn
-                  for turn in turns),
-              "the degenerate rejection receipt reached the model as its next turn")
-
-        print("== SSE ==")
-        run_session = db_query(
-            "select sc.session_id from session_commands sc join sessions s on s.id = sc.session_id"
-            " where s.task_id = ? and sc.type = 'plan' order by sc.id limit 1",
-            (run_task,))[0][0]
+        print("== D-side through both binaries: tool results + loop warning ==")
+        messages = capture_messages()
+        tool_turns = [text for role, text in messages if role == "tool"]
+        user_turns = [text for role, text in messages if role == "user"]
+        check(any("probing-the-workspace" in turn for turn in tool_turns),
+              "the probes' tool results reached the model as role=tool turns")
+        check(any(ephemeral in turn for turn in tool_turns),
+              "the branch probe's result (the ephemeral branch name) reached the model")
+        check(any("you ran the exact same command" in turn and "you are in a loop" in turn
+                  for turn in user_turns),
+              "the loop-warning notice reached the model as its next user turn")
+        check(any("Human selected" in turn for turn in user_turns),
+              "the question's verdict wording reached the model on resume")
+
+        print("== event log page + SSE ==")
+        # The §7.4 JSON page: newest-first window of the run task's
+        # ledger (id, session_id, seq, kind, actor, payload).
+        status, page = saddle_api("GET", "/tasks/%d/events?limit=200" % run_task, None)
+        page_events = page.get("events", []) if status == 200 else []
+        check(status == 200 and len(page_events) >= 10,
+              "the event log page serves the run's ledger (%d events)" % len(page_events))
+        ids = [e["id"] for e in page_events]
+        check(ids == sorted(ids, reverse=True), "the page is newest-first")
+        check(any(e["kind"] == "tool_call" and e["payload"].get("tool") == "submit_plan"
+                  for e in page_events),
+              "turn cards render: submit_plan tool_call events in the page")
+        check(any(e["kind"] == "tool_result" and e["payload"].get("tool") == "run_command"
+                  and "stdout" in e["payload"].get("result", "") for e in page_events),
+              "tool_result events carry the §7.1 result shape (stdout embedded)")
+        check(any(e["kind"] == "turn_text" and "Reading the brief" in e["payload"].get("text", "")
+                  for e in page_events),
+              "the narrated turn renders as turn_text")
+        check(all(e.get("seq") is not None and e["seq"] > 0 for e in page_events
+                  if e["kind"] in ("turn_text", "tool_call", "tool_result", "usage")),
+              "daemon events carry their per-session seq")
         state = goal_state(auth, goal1)
-        check(any(r.get("terminal_url", "").endswith("/terminal?session_id=%d" % run_session)
+        check(any(r.get("terminal_url", "").endswith("/tasks/%d/log" % run_task)
                   for r in state.get("runs", [])),
-              "the goal page's run list deep-links each session terminal")
-        request = urllib.request.Request(WEB + "/terminal-stream/%d" % run_session)
+              "the goal page's run list deep-links each session's event log")
+        request = urllib.request.Request(WEB + "/tasks/%d/events/stream" % run_task)
         with urllib.request.urlopen(request, timeout=15) as stream:
             check(stream.headers.get("Content-Type", "").startswith("text/event-stream"),
-                  "terminal stream is text/event-stream")
-            entries = []
+                  "event stream is text/event-stream")
+            streamed = []
             deadline = time.monotonic() + 10
-            while time.monotonic() < deadline and len(entries) < 120:
-                if any(e.get("type") == "done" for e in entries):
-                    break
+            while time.monotonic() < deadline and len(streamed) < 120:
                 line = stream.readline().decode().strip()
                 if not line.startswith("data: ") or line == "data: ping":
                     continue
-                entries.append(json.loads(line[6:]))
-            check(len(entries) >= 4, "terminal stream replays the run's backlog (%d entries)" % len(entries))
-            # push-era sessions stream their raw PTY log (the typed
-            # commands live in session_commands) — the plan submission
-            # shows as its echoed heredoc in the raw replay
-            check(any("saddle plan" in json.dumps(e) for e in entries),
-                  "terminal stream replay carries the plan submission round")
+                try:
+                    streamed.append(json.loads(line[6:]))
+                except ValueError:
+                    continue
+                if len(streamed) >= len(page_events):
+                    break
+            check(len(streamed) >= 10,
+                  "the stream replays the initial most-recent-30 window (%d events)" % len(streamed))
+            check(any(e.get("kind") == "tool_call" for e in streamed)
+                  and any(e.get("kind") == "tool_result" for e in streamed),
+                  "the SSE replay carries tool_call/tool_result events from the daemon")
 
         pulse = directives("/goals/%d" % goal1, auth)
         check(any(d.get("type") == "subscribe" for d in pulse),
               "goal page subscribes to the pulse channel for live refresh")
 
-        print("== push wire surface (the poll-era endpoints deleted) ==")
+        print("== push wire surface (§7.2 chunks + results states) ==")
         saddles_state = slot_public(auth, "/saddles", "saddles")
         act(auth, saddles_state["register"], {"name": "saddle-e2e-api", "host": "localhost:7199"})
         api_rows = [s for s in slot_public(auth, "/saddles", "saddles")["saddles"]
@@ -862,7 +911,7 @@ def main():
         act(auth, tasks_state["create"], {
             "project_id": "1",
             "title": "API surface probe",
-            "description": "Typed entries, exactly-once pushes",
+            "description": "Chunks ingestion, exactly-once pushes",
             "priority": "6",
             "task_branch": "task/api-surface",
         })
@@ -876,28 +925,36 @@ def main():
         api_session = db.execute("select max(id) from sessions").fetchone()[0]
         db.close()
 
-        def push(entry_id, entry_type, state="working", command="", output="",
-                 block=None, **extra):
-            """One round-results push in the daemon's shape (the
-            push-era wire the real daemon speaks — the poll-era
-            /update channel is gone, 404-pinned below)."""
+        def push_state(state="working", block=None, summary=None, entries=None):
+            """One state push in the daemon's /results shape."""
             payload = {"session_id": api_session, "state": state}
-            if entry_type is not None:
-                entry = {"id": entry_id, "command": command, "output": output,
-                         "exit_code": 0, "type": entry_type,
-                         "timestamp": "2026-09-13T00:00:00Z"}
-                entry.update(extra)
-                payload["entries"] = [entry]
             if block is not None:
                 payload["block"] = block
+            if summary is not None:
+                payload["summary"] = summary
+            if entries is not None:
+                payload["entries"] = entries
             return saddle_api("POST", "/api/saddle/results", api_token, payload)
 
+        def push_events(events):
+            """One §7.2 chunks batch ({task_id, session_id, events})."""
+            return saddle_api("POST", "/api/saddle/chunks", api_token,
+                              {"task_id": api_task, "session_id": api_session,
+                               "events": events})
+
         status, _ = saddle_api("POST", "/api/saddle/results", None,
                                {"session_id": api_session, "state": "working"})
         check(status == 401, "results push without a token is rejected")
         status, _ = saddle_api("POST", "/api/saddle/results", "saddle-not-a-token",
                                {"session_id": api_session, "state": "working"})
         check(status == 401, "results push with an invalid token is rejected")
+        status, _ = saddle_api("POST", "/api/saddle/chunks", None,
+                               {"session_id": api_session, "events": []})
+        check(status == 401, "chunks push without a token is rejected")
+        status, _ = saddle_api("POST", "/api/saddle/chunks", api_token,
+                               {"session_id": api_session, "events": [
+                                   {"kind": "progress", "actor": "model", "ts": "now"}]})
+        check(status == 400, "an event without a positive seq is refused (seq is mandatory)")
 
         # The poll-era ingest pair is GONE (arch 19): both routes 404
         # for any caller, authenticated or not — the subtask/plan
@@ -919,13 +976,39 @@ def main():
         check(status == 404,
               "the /update plan arm is gone with the channel (404, plans are goal-only)")
 
-        status, body = push(9001, "comment", command="saddle comment",
-                            output="Surveyed the module")
-        check(status == 200 and body.get("ok"), "comment entry accepted on the results push")
-        check(any(c.get("from_agent") and c["body"] == "Surveyed the module"
-                  for c in task_state(auth, api_task)["comments"]),
-              "the comment entry authors an agent row (output-first)")
-        status, body = push(None, None, state="blocked", block={
+        # The retired results-entries array is consumed silently: no
+        # comment row, no projection — round entries live as ledger
+        # events on the chunks channel now.
+        status, body = push_state(entries=[{"id": 9001, "command": "saddle comment",
+                                            "output": "Surveyed the module", "exit_code": 0,
+                                            "type": "comment",
+                                            "timestamp": "2026-09-13T00:00:00Z"}])
+        check(status == 200 and body.get("ok"),
+              "a results push carrying entries is accepted (consumed silently)")
+        check(not any(c["body"] == "Surveyed the module"
+                      for c in task_state(auth, api_task)["comments"]),
+              "the retired entries array projects nothing (no comment row)")
+
+        # §7.2 ingestion: a progress event lands in the ledger and the
+        # task's progress view; the (session_id, seq) unique key makes
+        # redelivery exactly-once.
+        status, body = push_events([{"seq": 9000, "kind": "progress", "actor": "model",
+                                     "ts": "2026-09-13T00:00:00Z",
+                                     "payload": {"text": "Migrated the schemas"}}])
+        check(status == 200 and body.get("stored") == 1,
+              "the chunks push stores the progress event")
+        status, body = push_events([{"seq": 9000, "kind": "progress", "actor": "model",
+                                     "ts": "2026-09-13T00:00:01Z",
+                                     "payload": {"text": "Migrated the schemas AGAIN"}}])
+        check(status == 200 and body.get("stored") == 0,
+              "the redelivered seq is consumed silently (UNIQUE(session_id, seq))")
+        progress_view = task_state(auth, api_task)["progress"]
+        check(len([p for p in progress_view
+                   if p["summary"] in ("Migrated the schemas", "Migrated the schemas AGAIN")]) == 1
+              and progress_view[0]["summary"] == "Migrated the schemas",
+              "progress events project exactly once per seq")
+
+        status, body = push_state(state="blocked", block={
             "question": "Which database?", "type": "question",
             "options": ["SQLite", "Postgres"], "body": "# Pick one"})
         check(status == 200 and body.get("ok"),
@@ -933,38 +1016,19 @@ def main():
         parked = db_query("select outcome from sessions where id = ?", (api_session,))[0][0]
         check(parked == "parked",
               "the raise parked the round's session (outcome 'parked', D4)")
-        status, body = push(9002, "query", command="cat notes.md | saddle query 'meaning?'",
-                            output="42", input="# notes\nmeaning of life")
-        check(status == 200, "query entry accepted on the results push")
-        query_rows = db_query(
-            "select sc.type, sc.piped_input from session_commands sc"
-            " where sc.session_id = ? and sc.command_text like '%saddle query%'",
-            (api_session,))
-        check(query_rows and query_rows[0][0] == "query"
-              and query_rows[0][1] == "# notes\nmeaning of life",
-              "the query entry persists typed with its piped input")
-        status, body = push(9003, "progress", command="saddle progress-update",
-                            output="Migrated the schemas")
-        check(status == 200, "progress entry accepted on the results push")
-        status, body = push(9003, "progress", command="saddle progress-update",
-                            output="Migrated the schemas AGAIN")
-        check(status == 200, "the redelivered entry id is consumed silently (200, no projection)")
-        check(len([p for p in task_state(auth, api_task)["progress"]
-                   if p["summary"] in ("Migrated the schemas", "Migrated the schemas AGAIN")])
-              == 1 and task_state(auth, api_task)["progress"][0]["summary"] == "Migrated the schemas",
-              "entries project exactly once per daemon-global id (push-era exactly-once)")
-        status, body = push(9004, "review", command="saddle review",
-                            output="Migration **ready** for review")
-        check(status == 200, "review entry accepted on the results push")
+        status, body = push_state(state="blocked", block={
+            "question": "Review the migration", "type": "review",
+            "body": "Migration **ready** for review"})
+        check(status == 200, "the review-type block push is accepted")
 
         detail = task_state(auth, api_task)
-        check(detail["status"] == "review_requested", "the review entry flips the task status")
+        check(detail["status"] == "review_requested", "the review block flips the task status")
         check([b["options"] for b in detail["blocks"]
                if b["block_type"] == "question"] == [["SQLite", "Postgres"]],
               "the pushed block carries its options")
         check(detail["progress"][0]["summary"] == "Migrated the schemas",
               "progress history newest-first")
-        status, _ = push(None, None, state="working")
+        status, _ = push_state(state="working")
         check(status == 200 and task_state(auth, api_task)["status"] == "review_requested",
               "a park-unwinding working push never resurrects the parked round")