C3: goal_events deleted — terminal watermark + derived narrative

goal_events is gone (M0001 amended) and nothing writes narrative rows
on ingest any more. goals.last_seen_event_id becomes
last_terminal_watermark (a timestamp): resume_due fires when the
goal's worker tasks carry terminal transitions (completed_at) newer
than the watermark and the oldest has aged past the coalescing
window; a paused goal simply queues them (status gate) until resume.
take_resume_notice keeps its shape (statuses + new completion
reports) and advances the watermark past every transition visible at
consume time.

The supervision narrative is now a DERIVED read-time view
(GoalService.narrative, newest first) over the goal's lifecycle
stamps, the run task's blocks (plan/completion submissions and their
verdicts, questions/RFCs) and its ONE comment thread (agent/user
footnotes, harness rows as system events), and the goal's tasks'
created/terminal transitions — the failure CASCADE derives from the
cancelled tasks surviving under a failed goal. Timestamp-less
bookkeeping never returns: task-enqueued/task-deleted/resumed are
gone, retry/requeue land as run-thread system rows. The goal page,
the goals list and the dashboard timeline (goal-* kinds, footnote
weights) all render from the derivation; plan-approval
context_updates key on the answered plan block instead of an event
row. The timeline's footnote set gains goal-system.

Tests: narrative content equivalence on the recorded run shape,
watermark no-refire after the pause flush, goal_events absence, and
the wake fixtures arm terminal task rows (fresh stamps) instead of
event inserts.

src/Api/SaddleApi.vala

@@ -741,9 +741,6 @@ namespace ClankerSaddle.Api {
                 foreach (var capability in resolved) {
                     yield capabilities.attach_to_task(capability.id, task.id);
                 }
-                yield goals.record_event(goal_id, "task-created",
-                    @"Task #$(task.id) $(task.title) created by the goal agent ($(mode))",
-                    task.id);
                 var response = new Json.Object();
                 response.set_int_member("task_id", task.id);
                 return SaddleUpdateEndpoint.json_ok(response);
@@ -787,8 +784,6 @@ namespace ClankerSaddle.Api {
                         "Only draft tasks can be deleted (cancel handles the rest)");
                 }
                 yield tasks.delete_task(task_id);
-                yield goals.record_event(goal_id, "task-deleted",
-                    @"Task #$(task_id) deleted by the goal agent", task_id);
                 return SaddleUpdateEndpoint.json_ok(new Json.Object());
             } catch (Error e) {
                 return SaddleUpdateEndpoint.json_error(400, e.message);
@@ -828,9 +823,6 @@ namespace ClankerSaddle.Api {
                         @"Task $(task_id) does not belong to your goal");
                 }
                 yield tasks.enqueue_task(task_id, CREATOR_GOAL);
-                yield goals.record_event(goal_id, "task-enqueued",
-                    @"Task #$(task_id) $(((!)detail).task.title) enqueued by the goal agent",
-                    task_id);
                 return SaddleUpdateEndpoint.json_ok(new Json.Object());
             } catch (Error e) {
                 return SaddleUpdateEndpoint.json_error(400, e.message);

src/Data/Entities.vala

@@ -547,7 +547,14 @@ namespace ClankerSaddle.Data {
         public string? plan { get; set; default = null; }
         public string? completion_report { get; set; default = null; }
         public string creator { get; set; default = "human"; }
-        public int64 last_seen_event_id { get; set; default = 0; }
+        /**
+         * Plan 15 C3 (was last_seen_event_id): the timestamp through
+         * which the goal agent has consumed its tasks' terminal
+         * transitions — resume_due fires on transitions newer than
+         * it, and a resume advance it. The narrative itself is
+         * derived at read time.
+         */
+        public DateTime? last_terminal_watermark { get; set; default = null; }
         public string? branch_push_state { get; set; default = null; }
         public DateTime? created_at { get; set; default = null; }
         public DateTime? started_at { get; set; default = null; }
@@ -567,7 +574,7 @@ namespace ClankerSaddle.Data {
                 .column<string>("plan", o => o.plan, (o, v) => o.plan = v)
                 .column<string>("completion_report", o => o.completion_report, (o, v) => o.completion_report = v)
                 .column<string>("creator", o => o.creator, (o, v) => o.creator = v)
-                .column<int64?>("last_seen_event_id", o => o.last_seen_event_id, (o, v) => o.last_seen_event_id = v ?? 0)
+                .column<DateTime?>("last_terminal_watermark", o => o.last_terminal_watermark, (o, v) => o.last_terminal_watermark = v)
                 .column<string>("branch_push_state", o => o.branch_push_state, (o, v) => o.branch_push_state = v)
                 .column<DateTime?>("created_at", o => o.created_at, (o, v) => o.created_at = v)
                 .column<DateTime?>("started_at", o => o.started_at, (o, v) => o.started_at = v)
@@ -593,30 +600,4 @@ namespace ClankerSaddle.Data {
         }
     }
 
-    /**
-     * One entry of a goal's supervision narrative (`goal_events`):
-     * lifecycle moves, plan submissions/reviews, the goal agent's task
-     * operations and its tasks' terminal events. Also the resume
-     * machinery's queue — {@link Goal.last_seen_event_id} marks the
-     * events the goal agent has already consumed.
-     */
-    public class GoalEvent : Object {
-
-        public int64 id { get; set; }
-        public int64 goal_id { get; set; }
-        public string kind { get; set; default = ""; }
-        public string text { get; set; default = ""; }
-        public int64 task_id { get; set; default = 0; }
-        public DateTime? created_at { get; set; default = null; }
-
-        public static void entity_mapping(EntityMapperBuilder<GoalEvent> cfg) {
-            cfg.table("goal_events")
-                .column<int64?>("id", o => o.id, (o, v) => o.id = v)
-                .column<int64?>("goal_id", o => o.goal_id, (o, v) => o.goal_id = v)
-                .column<string>("kind", o => o.kind, (o, v) => o.kind = v)
-                .column<string>("text", o => o.text, (o, v) => o.text = v)
-                .column<int64?>("task_id", o => o.task_id, (o, v) => o.task_id = v ?? 0)
-                .column<DateTime?>("created_at", o => o.created_at, (o, v) => o.created_at = v);
-        }
-    }
 }

src/Data/Migrations/M0001_Initial.vala

@@ -22,7 +22,9 @@ namespace ClankerSaddle.Data.Migrations {
      * notes; capability tags; and the plan-11 notification tables for
      * task E
      * (`push_subscriptions`, `task_notify_prefs` — included now so
-     * D+E needs no migration).
+     * D+E needs no migration). Plan 15 C3 deleted `goal_events` (the
+     * narrative is derived; the terminal-transition watermark lives
+     * on the goals row).
      */
     public class M0001_Initial : Migration {
 
@@ -64,10 +66,14 @@ namespace ClankerSaddle.Data.Migrations {
             // user-named goal branch (uniqueness-validated against the
             // other live goals sharing involved repos); `plan` holds
             // the APPROVED plan markdown (rejected revisions live on
-            // as the run task's answered plan blocks); `last_seen_event_id`
-            // is the goal agent's consumed-event watermark (resume
-            // coalescing); `branch_push_state` caches the
-            // daemon-reported push state of the goal branches.
+            // as the run task's answered plan blocks);
+            // `last_terminal_watermark` (plan 15 C3, replacing the
+            // goal_events-id watermark) is the timestamp through which
+            // the goal agent has consumed its tasks' TERMINAL
+            // transitions (resume_due computes from transitions newer
+            // than it; the narrative itself is a derived read-time
+            // view); `branch_push_state` caches the daemon-reported
+            // push state of the goal branches.
             b.create_table("goals", (t) => {
                 t.column<int64?>("id")
                     .auto_increment()
@@ -92,8 +98,7 @@ namespace ClankerSaddle.Data.Migrations {
                 t.column<string>("completion_report");
                 t.column<string>("creator")
                     .default_value<string>("human");
-                t.column<int64?>("last_seen_event_id")
-                    .default_value<int64?>(0);
+                t.column<DateTime?>("last_terminal_watermark");
                 t.column<string>("branch_push_state");
                 t.column<DateTime?>("created_at");
                 t.column<DateTime?>("started_at");
@@ -219,26 +224,13 @@ namespace ClankerSaddle.Data.Migrations {
             // the run task's task_comments (human comments inject
             // through the existing delivery paths).
 
-            // The supervision narrative's event log: one row per
-            // notable goal occurrence (plan reviews, agent task
-            // operations, task terminal events, lifecycle moves) —
-            // the resume machinery's watermark (`goals.
-            // last_seen_event_id`) also lives on these ids.
-            b.create_table("goal_events", (t) => {
-                t.column<int64?>("id")
-                    .auto_increment()
-                    .primary_key();
-                t.column<int64?>("goal_id")
-                    .not_null()
-                    .references("goals", "id")
-                    .indexed();
-                t.column<string>("kind")
-                    .not_null();
-                t.column<string>("text")
-                    .not_null();
-                t.column<int64?>("task_id");
-                t.column<DateTime?>("created_at");
-            });
+            // Plan 15 C3: the goal_events table is GONE — the
+            // supervision narrative is a DERIVED read-time view (over
+            // the goal's lifecycle stamps, the run task's blocks and
+            // comments, and the goal's tasks' created/terminal
+            // transitions), and the resume watermark lives on the
+            // goals row as last_terminal_watermark. Nothing is
+            // written on ingest.
 
             b.create_table("saddles", (t) => {
                 t.column<int64?>("id")
@@ -499,7 +491,6 @@ namespace ClankerSaddle.Data.Migrations {
 
         public override void down(MigrationBuilder b) {
             b.drop_table("saddle_entry_dedup");
-            b.drop_table("goal_events");
             b.drop_table("task_notify_prefs");
             b.drop_table("push_subscriptions");
             b.drop_table("task_capabilities");

src/Services/AllocationService.vala

@@ -467,17 +467,24 @@ namespace ClankerSaddle.Services {
 
             var report = new StringBuilder();
 
-            // Plan approval (goal work only): the plan-approved event
+            // Plan approval (goal work only): an approved plan BLOCK
+            // (plan 15 C3 — the answered run-task row IS the record)
             // after the watermark names the round that missed it.
             if (task.goal_id > 0) {
                 bool approved = false;
-                foreach (var goal_event in yield db.query<GoalEvent>()
-                        .where(expr("goal_id == $0 && kind == $1",
-                            elem<int64?>(task.goal_id), elem("plan-approved")))
-                        .materialise_async()) {
-                    if (goal_event.created_at != null
-                            && ((!)goal_event.created_at).compare((!)watermark) > 0) {
-                        approved = true;
+                var approval_run = yield db.query<TaskEntity>()
+                    .where(expr("goal_id == $0 && kind == $1",
+                        elem<int64?>(task.goal_id), elem(TASK_KIND_GOAL_RUN)))
+                    .first_async();
+                if (approval_run != null) {
+                    foreach (var block in yield tasks.list_blocks(
+                            ((!)approval_run).id)) {
+                        if (block.block_type == BLOCK_PLAN
+                                && block.answer == NOTICE_PLAN_APPROVED
+                                && block.answered_at != null
+                                && ((!)block.answered_at).compare((!)watermark) > 0) {
+                            approved = true;
+                        }
                     }
                 }
                 if (approved) {

src/Services/GoalService.vala

@@ -142,7 +142,6 @@ namespace ClankerSaddle.Services {
                 created_at = new DateTime.now_local()
             };
             yield db.insert_async<Goal>(goal);
-            yield record_event(goal.id, "created", @"Goal created (branch $(trimmed_branch))");
             yield pulse.bump(0, "goal created");
             return goal;
         }
@@ -270,8 +269,6 @@ namespace ClankerSaddle.Services {
                 null, ((!)goal).base_branch, STATUS_QUEUED,
                 TASK_KIND_GOAL_RUN, MODE_ORCHESTRATE, CREATOR_GOAL,
                 ((!)goal).id, 0);
-            yield record_event(goal_id, "started",
-                @"Goal started — the goal agent is planning (run task #$(run.id))", run.id);
             yield pulse.bump(run.id, "goal started");
             return goal;
         }
@@ -290,7 +287,6 @@ namespace ClankerSaddle.Services {
             ((!)goal).status = GOAL_PAUSED;
             ((!)goal).paused_at = new DateTime.now_local();
             yield db.update_async<Goal>((!)goal);
-            yield record_event(goal_id, "paused", "Goal paused — supervision frozen (events queue)");
             yield pulse.bump(0, "goal paused");
             return goal;
         }
@@ -309,7 +305,6 @@ namespace ClankerSaddle.Services {
             ((!)goal).status = GOAL_ACTIVE;
             ((!)goal).paused_at = null;
             yield db.update_async<Goal>((!)goal);
-            yield record_event(goal_id, "resumed", "Goal resumed — queued events coalesce into the next agent wake");
             yield pulse.bump(0, "goal resumed");
             return goal;
         }
@@ -347,7 +342,6 @@ namespace ClankerSaddle.Services {
                     yield tasks.cancel_task(task.id);
                 }
             }
-            yield record_event(goal_id, "cancelled", "Goal cancelled — its tasks were cancelled with it");
             yield pulse.bump(0, "goal cancelled");
         }
 
@@ -384,8 +378,6 @@ namespace ClankerSaddle.Services {
             }
             ((!)goal).status = GOAL_PLAN_REVIEW;
             yield db.update_async<Goal>(((!)goal));
-            yield record_event(goal_id, "plan-submitted",
-                @"Plan revision #$(((!)raised).id) submitted for review");
             return raised;
         }
 
@@ -404,8 +396,6 @@ namespace ClankerSaddle.Services {
             if (raised == null) {
                 return null;
             }
-            yield record_event(goal_id, "block-raised",
-                @"$(((!)raised).block_type) raised: $(question)");
             return raised;
         }
 
@@ -437,8 +427,6 @@ namespace ClankerSaddle.Services {
             ((!)goal).status = GOAL_COMPLETION_REVIEW;
             ((!)goal).completion_report = report;
             yield db.update_async<Goal>(((!)goal));
-            yield record_event(goal_id, "completion-submitted",
-                "Completion report submitted for review");
             return raised;
         }
 
@@ -490,11 +478,7 @@ namespace ClankerSaddle.Services {
                 }
                 cascaded++;
             }
-            yield record_event(goal_id, "goal-failed",
-                report != null && report.length > 0 ? report : "The goal agent failed the goal");
             if (cascaded > 0) {
-                yield record_event(goal_id, "cascade",
-                    @"Goal failure cancelled $(cascaded) outstanding task(s)");
             }
             push.notify_global(PushService.goal_terminal_notification(
                 ((!)goal).title, goal_id, false));
@@ -560,23 +544,15 @@ namespace ClankerSaddle.Services {
                         ((!)goal).plan = ((!)block).body;
                     }
                     ((!)goal).status = GOAL_ACTIVE;
-                    yield record_event(((!)goal).id, "plan-approved",
-                        @"Plan revision #$(((!)block).id) approved — the goal is active");
                 } else {
                     ((!)goal).status = GOAL_PLANNING;
-                    yield record_event(((!)goal).id, "plan-rejected",
-                        @"Plan revision #$(((!)block).id) rejected with feedback — the agent revises");
                 }
             } else if (((!)block).block_type == GOAL_BLOCK_COMPLETION) {
                 if (answer == NOTICE_REVIEW_APPROVED) {
                     ((!)goal).status = GOAL_COMPLETE;
                     ((!)goal).completed_at = new DateTime.now_local();
-                    yield record_event(((!)goal).id, "goal-approved",
-                        "Completion report approved — the goal is complete");
                 } else {
                     ((!)goal).status = GOAL_ACTIVE;
-                    yield record_event(((!)goal).id, "completion-rejected",
-                        "Completion report rejected with feedback — the agent continues");
                 }
             }
             yield db.update_async<Goal>(((!)goal));
@@ -686,31 +662,206 @@ namespace ClankerSaddle.Services {
             return yield tasks.list_blocks(((!)run).id);
         }
 
-        /** A goal's narrative events, oldest first. */
-        public async ImmutableLot<GoalEvent> list_events(int64 goal_id) throws Error {
-            return yield db.query<GoalEvent>()
-                .where(expr("goal_id == $0", elem<int64?>(goal_id)))
-                .order_by(expr("id"))
-                .materialise_async();
+        /**
+         * One derived supervision-narrative row (plan 15 C3): the
+         * read-time projection of a timestamped fact — a lifecycle
+         * stamp, a run-task block or comment, a worker task's
+         * created/terminal transition. `kind` keeps the old
+         * goal_events vocabulary; system rows (kind "system") come
+         * from the run thread's harness-authored comments.
+         */
+        public class NarrativeRow : Object {
+            public string kind { get; set; default = ""; }
+            public string text { get; set; default = ""; }
+            public int64 task_id { get; set; }
+            public DateTime? at { get; set; default = null; }
         }
 
         /**
-         * Records one supervision-narrative event. Called by the
-         * lifecycle flows, the goal agent's task operations and the
-         * task-terminal hook — every goal_events row is also a
-         * potential resume signal (see {@link resume_due}).
+         * The supervision narrative (plan 15 C3): a DERIVED read-time
+         * view — nothing is written on ingest any more. The rows come
+         * from the goal's lifecycle stamps (created/started/paused/
+         * terminal), the goal-run task's blocks (plan submissions and
+         * verdicts, questions/RFCs, the completion review) and its
+         * comment thread (agent/user rows as footnote comments,
+         * harness-authored rows as system events), and the goal's
+         * tasks' created and terminal transitions. Timestamp-less
+         * bookkeeping (task-enqueued, task-deleted, resumed) never
+         * returns; the goal-failure CASCADE derives from the
+         * cancelled tasks surviving under a failed goal. Newest
+         * first.
          */
-        public async GoalEvent record_event(int64 goal_id, string kind, string text,
-                int64 task_id = 0, int64 ref_task_id = 0) throws Error {
-            var event = new GoalEvent() {
-                goal_id = goal_id,
-                kind = kind,
-                text = text,
-                task_id = task_id,
-                created_at = new DateTime.now_local()
-            };
-            yield db.insert_async<GoalEvent>(event);
-            return event;
+        public async ImmutableLot<NarrativeRow> narrative(int64 goal_id) throws Error {
+            var goal = yield get_goal(goal_id);
+            if (goal == null) {
+                return new Vector<NarrativeRow>().to_immutable_buffer();
+            }
+            var rows = new Vector<NarrativeRow>();
+
+            // Lifecycle stamps.
+            if (((!)goal).created_at != null) {
+                rows.add(new NarrativeRow() {
+                    kind = "created", at = ((!)goal).created_at,
+                    text = @"Goal created (branch $(((!)goal).branch_name))" });
+            }
+            if (((!)goal).started_at != null) {
+                rows.add(new NarrativeRow() {
+                    kind = "started", at = ((!)goal).started_at,
+                    text = "Goal started — the goal agent is planning" });
+            }
+            if (((!)goal).paused_at != null && ((!)goal).status == GOAL_PAUSED) {
+                rows.add(new NarrativeRow() {
+                    kind = "paused", at = ((!)goal).paused_at,
+                    text = "Goal paused — supervision frozen (terminal transitions queue)" });
+            }
+            if (((!)goal).completed_at != null) {
+                if (((!)goal).status == GOAL_CANCELLED) {
+                    rows.add(new NarrativeRow() {
+                        kind = "cancelled", at = ((!)goal).completed_at,
+                        text = "Goal cancelled — its tasks were cancelled with it" });
+                } else if (((!)goal).status == GOAL_FAILED) {
+                    rows.add(new NarrativeRow() {
+                        kind = "goal-failed", at = ((!)goal).completed_at,
+                        text = (((!)goal).completion_report ?? "").length > 0
+                            ? (!)(((!)goal).completion_report)
+                            : "The goal agent failed the goal" });
+                }
+            }
+
+            // The run task's blocks: the plan/completion reviews with
+            // their verdicts, the questions/RFCs.
+            var run = yield goal_run_task(goal_id);
+            if (run != null) {
+                foreach (var block in yield tasks.list_blocks(((!)run).id)) {
+                    if (block.created_at != null) {
+                        if (block.block_type == GOAL_BLOCK_PLAN) {
+                            rows.add(new NarrativeRow() {
+                                kind = "plan-submitted", at = block.created_at,
+                                text = @"Plan revision #$(block.id) submitted for review" });
+                        } else if (block.block_type == GOAL_BLOCK_COMPLETION) {
+                            rows.add(new NarrativeRow() {
+                                kind = "completion-submitted", at = block.created_at,
+                                text = "Completion report submitted for review" });
+                        } else {
+                            rows.add(new NarrativeRow() {
+                                kind = "block-raised", at = block.created_at,
+                                text = @"$(block.block_type) raised: $(block.question)" });
+                        }
+                    }
+                    if (block.answered_at != null) {
+                        if (block.block_type == GOAL_BLOCK_PLAN) {
+                            bool approved = block.answer == NOTICE_PLAN_APPROVED;
+                            rows.add(new NarrativeRow() {
+                                kind = approved ? "plan-approved" : "plan-rejected",
+                                at = block.answered_at,
+                                text = approved
+                                    ? @"Plan revision #$(block.id) approved — the goal is active"
+                                    : @"Plan revision #$(block.id) rejected with feedback — the agent revises" });
+                        } else if (block.block_type == GOAL_BLOCK_COMPLETION) {
+                            bool approved = block.answer == NOTICE_REVIEW_APPROVED;
+                            rows.add(new NarrativeRow() {
+                                kind = approved ? "goal-approved" : "completion-rejected",
+                                at = block.answered_at,
+                                text = approved
+                                    ? "Completion report approved — the goal is complete"
+                                    : "Completion report rejected with feedback — the agent continues" });
+                        } else {
+                            rows.add(new NarrativeRow() {
+                                kind = "block-answered", at = block.answered_at,
+                                text = @"$(block.block_type) answered" });
+                        }
+                    }
+                }
+                // The ONE comment thread: agent/user rows as footnote
+                // comments, harness-authored rows as system events.
+                foreach (var comment in yield tasks.list_comments(((!)run).id)) {
+                    if (comment.created_at == null) {
+                        continue;
+                    }
+                    rows.add(new NarrativeRow() {
+                        kind = comment.author == COMMENT_AUTHOR_SYSTEM
+                            ? "system" : "comment",
+                        at = comment.created_at,
+                        text = comment.body });
+                }
+            }
+
+            // The goal's tasks: created + terminal transitions. The
+            // failure cascade derives from the cancelled tasks
+            // surviving under a failed goal (deleted drafts left no
+            // rows; the count is the surviving record).
+            int64 cascaded = 0;
+            foreach (var task in yield db.query<TaskEntity>()
+                    .where(expr("goal_id == $0 && kind == $1",
+                        elem<int64?>(goal_id), elem(TASK_KIND_TASK)))
+                    .order_by(expr("id"))
+                    .materialise_async()) {
+                if (task.created_at != null) {
+                    rows.add(new NarrativeRow() {
+                        kind = "task-created", at = task.created_at, task_id = task.id,
+                        text = @"Task #$(task.id) $(task.title) created by the goal agent ($(task.mode))" });
+                }
+                if (task.completed_at != null) {
+                    if (task.status == STATUS_DONE) {
+                        rows.add(new NarrativeRow() {
+                            kind = "task-done", at = task.completed_at, task_id = task.id,
+                            text = @"Task #$(task.id) $(task.title) — completed$(summary_lead(task))" });
+                    } else if (task.status == STATUS_FAILED) {
+                        rows.add(new NarrativeRow() {
+                            kind = "task-failed", at = task.completed_at, task_id = task.id,
+                            text = @"Task #$(task.id) $(task.title) — failed$(summary_lead(task))" });
+                        if (((!)goal).status == GOAL_FAILED
+                                && (task.summary ?? "") == "Cancelled by user") {
+                            cascaded++;
+                        }
+                    }
+                }
+            }
+            if (((!)goal).status == GOAL_FAILED && cascaded > 0) {
+                rows.add(new NarrativeRow() {
+                    kind = "cascade", at = ((!)goal).completed_at,
+                    text = @"Goal failure cancelled $(cascaded) outstanding task(s)" });
+            }
+
+            // Newest first, stable for equal timestamps (insertion
+            // sort — the narratives are small).
+            var array = rows.to_array();
+            for (var i = 1; i < array.length; i++) {
+                var key = array[i];
+                var j = i - 1;
+                while (j >= 0 && narrative_after(key, array[j])) {
+                    array[j + 1] = array[j];
+                    j--;
+                }
+                array[j + 1] = key;
+            }
+            var newest_first = new Vector<NarrativeRow>();
+            foreach (var row in array) {
+                newest_first.add(row);
+            }
+            return newest_first.to_immutable_buffer();
+        }
+
+        /** True when row `a` sorts after (newer than) row `b`. */
+        private static bool narrative_after(NarrativeRow a, NarrativeRow b) {
+            if (a.at == null) {
+                return false;
+            }
+            if (b.at == null) {
+                return true;
+            }
+            return ((!)a.at).compare((!)b.at) > 0;
+        }
+
+        /** The task's summary as a narrative lead-in (": <first line>"). */
+        private static string summary_lead(TaskEntity task) {
+            if (task.summary == null || ((!)task.summary).length == 0) {
+                return "";
+            }
+            string summary = (!)task.summary;
+            int cut = summary.index_of("\n");
+            string lead = cut >= 0 ? summary.substring(0, cut) : summary;
+            return @": $(lead)";
         }
 
         // ------------------------------------------------------------------
@@ -720,10 +871,14 @@ namespace ClankerSaddle.Services {
         /**
          * True when the goal agent should be resumed: its run task is
          * parked (suspended), the goal is active (paused goals freeze
-         * supervision; review states park on blocks instead), and at
-         * least one task event has waited out the coalescing window
-         * (the OLDEST unconsumed event is old enough — bursts collapse
-         * into one resume).
+         * supervision — transitions newer than the watermark simply
+         * QUEUE until the goal resumes; review states park on blocks
+         * instead), and at least one owned task's TERMINAL transition
+         * (completed_at newer than {@link Goal.last_terminal_watermark})
+         * has waited out the coalescing window — the OLDEST such
+         * transition ages the window, so bursts collapse into one
+         * resume. Plan 15 C3: the watermark replaced the goal_events
+         * id cursor; the transitions read straight off the tasks.
          */
         public async bool resume_due(TaskEntity run_task) throws Error {
             if (run_task.kind != TASK_KIND_GOAL_RUN || run_task.status != STATUS_SUSPENDED) {
@@ -733,23 +888,25 @@ namespace ClankerSaddle.Services {
             if (goal == null || ((!)goal).status != GOAL_ACTIVE) {
                 return false;
             }
+            DateTime? watermark = ((!)goal).last_terminal_watermark;
             DateTime? oldest = null;
-            // ONLY owned-task terminal events resume the parked agent —
-            // the agent's own task-created/task-enqueued narrative (and
-            // human comments) must never wake it; the first terminal
-            // event ages the window and bursts coalesce behind it.
-            foreach (var event in yield db.query<GoalEvent>()
-                    .where(expr("goal_id == $0 && id > $1",
-                        elem<int64?>(run_task.goal_id),
-                        elem<int64?>(((!)goal).last_seen_event_id)))
+            // ONLY owned-task terminal transitions resume the parked
+            // agent — the agent's own task-created narrative (and
+            // human comments) must never wake it.
+            foreach (var task in yield db.query<TaskEntity>()
+                    .where(expr("goal_id == $0 && kind == $1",
+                        elem<int64?>(run_task.goal_id), elem(TASK_KIND_TASK)))
                     .order_by(expr("id"))
                     .materialise_async()) {
-                if (event.kind != "task-done" && event.kind != "task-failed") {
+                if (task.completed_at == null) {
                     continue;
                 }
-                if (event.created_at != null) {
-                    oldest = ((!)event).created_at;
-                    break;
+                if (watermark != null
+                        && ((!)task.completed_at).compare((!)watermark) <= 0) {
+                    continue;
+                }
+                if (oldest == null || ((!)task.completed_at).compare((!)oldest) < 0) {
+                    oldest = task.completed_at;
                 }
             }
             if (oldest == null) {
@@ -764,8 +921,11 @@ namespace ClankerSaddle.Services {
          * assignment's `resume_notice`): the current status of ALL of
          * the goal's tasks plus the full completion reports of the
          * tasks that finished since the agent parked. Consumes the
-         * pending events (the watermark advances), so the window
-         * starts fresh for the next park.
+         * pending transitions — the terminal watermark advances past
+         * every transition visible now (and past the clock, so a
+         * same-second park/consume pair cannot re-fire) — so the
+         * window starts fresh for the next park. Plan 15 C3: the
+         * watermark replaced the goal_events id cursor.
          */
         public async string take_resume_notice(int64 goal_id) throws Error {
             var goal = yield require_goal(goal_id);
@@ -778,7 +938,6 @@ namespace ClankerSaddle.Services {
             var report = new StringBuilder();
             report.append("Goal update — resuming your parked session:\n\n");
             report.append("## Task statuses\n");
-            int64 newest_event = ((!)goal).last_seen_event_id;
             var finished = new Vector<TaskEntity>();
             foreach (var task in yield db.query<TaskEntity>()
                     .where(expr("goal_id == $0 && kind == $1",
@@ -810,17 +969,18 @@ namespace ClankerSaddle.Services {
                 .append(" prune worktrees, create or enqueue follow-ups as needed, then")
                 .append(" `saddle suspend` again when you are parked.\n");
 
-            foreach (var event in yield db.query<GoalEvent>()
+            DateTime consumed = new DateTime.now_local();
+            foreach (var task in yield db.query<TaskEntity>()
                     .where(expr("goal_id == $0", elem<int64?>(goal_id)))
-                    .order_by_desc(expr("id"))
-                    .limit(1)
+                    .order_by(expr("id"))
                     .materialise_async()) {
-                newest_event = event.id;
-            }
-            if (newest_event != ((!)goal).last_seen_event_id) {
-                ((!)goal).last_seen_event_id = newest_event;
-                yield db.update_async<Goal>((!)goal);
+                if (task.completed_at != null
+                        && ((!)task.completed_at).compare(consumed) > 0) {
+                    consumed = (!)task.completed_at;
+                }
             }
+            ((!)goal).last_terminal_watermark = consumed;
+            yield db.update_async<Goal>(((!)goal));
             return report.str;
         }
 

src/Services/SaddleService.vala

@@ -1299,13 +1299,10 @@ namespace ClankerSaddle.Services {
                             body = strip_quotes(strip_verb(command, "saddle comment"));
                         }
                         if (body.length > 0) {
+                            // Plan 15 C2/C3: the run task's comment IS
+                            // the goal comment — one row, no mirror,
+                            // and the narrative derives it at read time.
                             yield tasks.add_comment(task_id, "agent", body);
-                            if (goal_id > 0) {
-                                // Plan 15 C2: the run task's comment
-                                // IS the goal comment — no mirror.
-                                yield goal_service.record_event(goal_id, "comment",
-                                    snippet_text(body));
-                            }
                         }
                         break;
                     case "plan":

src/Services/TaskService.vala

@@ -137,34 +137,6 @@ namespace ClankerSaddle.Services {
         private StatePulse pulse = inject<StatePulse>();
         private PushService push = inject<PushService>();
 
-        /**
-         * The goal's task-terminal hook (plan 12): a worker task under
-         * a goal records a supervision-narrative event — the parked
-         * goal agent's resume machinery coalesces on these rows (see
-         * {@link GoalService.resume_due}). Inline (not via
-         * GoalService) because GoalService depends on this service.
-         */
-        private async void note_goal_task_event(TaskEntity task, string outcome_word) throws Error {
-            if (task.goal_id <= 0 || task.kind != TASK_KIND_TASK) {
-                return;
-            }
-            string lead = "";
-            if (task.summary != null && ((!)task.summary).length > 0) {
-                int cut = ((!)task.summary).index_of("\n");
-                lead = cut >= 0 ? ((!)task.summary).substring(0, cut) : (!)task.summary;
-            }
-            string text = lead.length > 0
-                ? @"Task #$(task.id) $(task.title) — $(outcome_word): $(lead)"
-                : @"Task #$(task.id) $(task.title) — $(outcome_word)";
-            yield db.insert_async<GoalEvent>(new GoalEvent() {
-                goal_id = task.goal_id,
-                kind = task.status == STATUS_DONE ? "task-done" : "task-failed",
-                text = text,
-                task_id = task.id,
-                created_at = new DateTime.now_local()
-            });
-        }
-
         /**
          * Inserts a task (status 'draft' until its creator enqueues
          * it). `description`, `task_branch` and `branch_from` may be
@@ -1138,7 +1110,6 @@ namespace ClankerSaddle.Services {
             // Plan 12: the goal's narrative hears the terminal event
             // (a parked goal agent resumes on it), and a done
             // predecessor auto-enqueues its dependents.
-            yield note_goal_task_event(((!)task), "completed");
             yield enqueue_dependents_of(((!)task));
         }
 
@@ -1175,7 +1146,6 @@ namespace ClankerSaddle.Services {
             yield close_open_sessions(task_id, "stopped");
             yield free_saddles_running(task_id);
             yield pulse.bump(task_id, "task cancelled");
-            yield note_goal_task_event(((!)task), "cancelled");
         }
 
         /**
@@ -1218,7 +1188,6 @@ namespace ClankerSaddle.Services {
             // not — the cancellator already knows).
             push.notify_opted_in(task_id, PushService.lifecycle_notification(
                 ((!)task).title, task_id, "failed", summary));
-            yield note_goal_task_event(((!)task), "failed");
         }
 
         /**

src/Services/TimelineService.vala

@@ -32,9 +32,10 @@ namespace ClankerSaddle.Services {
      * raised and answered, comments (author + snippet), progress
      * updates, completions/failures and session dispatches — plus the
      * plan-12 goal events (plan submissions/reviews, task
-     * fan-out, lifecycle) — materialised per request from the
-     * existing tables (tasks, task_blocks, task_comments,
-     * task_progress, sessions, goal_events). Nothing is archived or
+     * fan-out, lifecycle — plan 15 C3: derived per goal by
+     * {@link GoalService.narrative}) — materialised per request from
+     * the existing tables (tasks, task_blocks, task_comments,
+     * task_progress, sessions, goals). Nothing is archived or
      * aggregated: at this fleet size a bounded newest-cut per source,
      * merged and re-sorted, is the whole job.
      */
@@ -47,6 +48,7 @@ namespace ClankerSaddle.Services {
         public const int SNIPPET_MAX = 160;
 
         private OrmSession db = inject<OrmSession>();
+        private GoalService goals = inject<GoalService>();
 
         /**
          * The recent event window, newest first. Each source
@@ -225,26 +227,37 @@ namespace ClankerSaddle.Services {
             // goal agent's task operations, task terminal events and
             // lifecycle moves — keyed on the goal (the timeline links
             // the goal page, not the task).
-            foreach (var goal_event in yield db.query<GoalEvent>()
-                    .order_by_desc(expr("id")).limit(WINDOW).materialise_async()) {
-                if (goal_event.created_at == null) {
-                    continue;
+            // Plan 15 C3: the goal narrative derives at read time —
+            // per goal, newest WINDOW rows, keyed on the goal (the
+            // timeline links the goal page, not the task). Narrative
+            // comment rows land as goal-comment footnotes, harness
+            // system rows as goal-system footnotes.
+            foreach (var goal in yield db.query<Goal>().materialise_async()) {
+                int64 taken_per_goal = 0;
+                foreach (var row in yield goals.narrative(goal.id)) {
+                    if (row.at == null || taken_per_goal >= WINDOW) {
+                        continue;
+                    }
+                    taken_per_goal++;
+                    // A narrative row may reference the goal agent's
+                    // own run task — the row stays goal-keyed, but the
+                    // internal task never becomes a clickable task
+                    // reference.
+                    int64 event_task_id = run_tasks.has(row.task_id.to_string())
+                        ? 0 : row.task_id;
+                    string kind = row.kind == "comment" ? "goal-comment"
+                        : row.kind == "system" ? "goal-system" : @"goal-$(row.kind)";
+                    events.add(new TimelineEvent() {
+                        kind = kind,
+                        at_epoch = ((!)row.at).to_unix(),
+                        task_id = event_task_id,
+                        task_title = event_task_id > 0
+                            ? titles.get_or_default(event_task_id.to_string()) ?? "" : "",
+                        goal_id = goal.id,
+                        goal_title = goal_titles.get_or_default(goal.id.to_string()) ?? "",
+                        text = snippet(row.text), sequence = sequence++
+                    });
                 }
-                // A goal event may reference the goal agent's own run
-                // task — the row stays goal-keyed, but the internal
-                // task never becomes a clickable task reference.
-                int64 event_task_id = run_tasks.has(goal_event.task_id.to_string())
-                    ? 0 : goal_event.task_id;
-                events.add(new TimelineEvent() {
-                    kind = @"goal-$(goal_event.kind)",
-                    at_epoch = ((!)goal_event.created_at).to_unix(),
-                    task_id = event_task_id,
-                    task_title = event_task_id > 0
-                        ? titles.get_or_default(event_task_id.to_string()) ?? "" : "",
-                    goal_id = goal_event.goal_id,
-                    goal_title = goal_titles.get_or_default(goal_event.goal_id.to_string()) ?? "",
-                    text = snippet(goal_event.text), sequence = sequence++
-                });
             }
 
             var sorted = events.sort((a, b) => {

src/actions/GoalActions.vala

@@ -232,9 +232,11 @@ namespace ClankerSaddle {
                 ((!)goal).status = GOAL_ACTIVE;
                 ((!)goal).completed_at = null;
                 yield db.update_async<Goal>((!)goal);
-                yield goals.record_event(goal_id, "retry",
-                    @"Supervision retried — the goal agent runs again (run task #$(((!)run).id))",
-                    ((!)run).id);
+                // Plan 15 C3: the retry is harness bookkeeping — a
+                // system row on the run thread (the narrative derives
+                // it at read time).
+                yield tasks.add_system_comment(((!)run).id,
+                    @"Supervision retried — the goal agent runs again (run task #$(((!)run).id))");
                 return directives().notify("success",
                         @"Goal '$(((!)goal).title)' — supervision requeued for another attempt")
                     .navigate(@"/goals/$(goal_id)");
@@ -280,9 +282,8 @@ namespace ClankerSaddle {
                 }
                 yield saddles.abort_live_session(tasks, ((!)run).id, "requeued");
                 yield tasks.requeue_task(((!)run).id);
-                yield goals.record_event(goal_id, "requeue",
-                    @"Run requeued — the daemon restores the agent's workspace on the next dispatch (run task #$(((!)run).id))",
-                    ((!)run).id);
+                yield tasks.add_system_comment(((!)run).id,
+                    @"Run requeued — the daemon restores the agent's workspace on the next dispatch (run task #$(((!)run).id))");
                 return directives().notify("success",
                         "Run requeued; the daemon restores the agent's context on the next dispatch")
                     .navigate(@"/goals/$(goal_id)");

src/entrypoints/GoalDetailEntrypoint.vala

@@ -69,9 +69,13 @@ namespace ClankerSaddle {
         public string when_display { get; set; default = ""; }
     }
 
-    /** One supervision-narrative event row. */
+    /**
+     * One supervision-narrative row (plan 15 C3: the DERIVED read-time
+     * view over the goal's lifecycle stamps, the run task's blocks
+     * and comments, and the goal's tasks' transitions — no
+     * goal_events rows exist).
+     */
     public class GoalEventView : Object {
-        public int64 id { get; set; }
         public string kind { get; set; default = ""; }
         public string text { get; set; default = ""; }
         public int64 task_id { get; set; }
@@ -527,21 +531,16 @@ namespace ClankerSaddle {
                 }
             }
 
-            var event_buffer = new Vector<GoalEventView>();
-            foreach (var row in yield goals.list_events(goal.id)) {
-                event_buffer.add(new GoalEventView() {
-                    id = row.id,
+            // Plan 15 C3: the narrative derives at read time (the
+            // service returns newest-first rows).
+            foreach (var row in yield goals.narrative(goal.id)) {
+                events.add(new GoalEventView() {
                     kind = row.kind,
                     text = row.text,
                     task_id = row.task_id,
-                    at_display = format_stamp(row.created_at)
+                    at_display = format_stamp(row.at)
                 });
             }
-            // Newest first: the narrative reads like a feed.
-            var event_array = event_buffer.to_array();
-            for (int i = event_array.length - 1; i >= 0; i--) {
-                events.add(event_array[i]);
-            }
 
             // Plan 15 C2: the goal's comment thread IS the run
             // task's task_comments (injection rides the existing

src/main.vala

@@ -50,7 +50,6 @@ int main(string[] args) {
         db.add_entity<TaskNotifyPref>(TaskNotifyPref.entity_mapping);
         db.add_entity<Goal>(Goal.entity_mapping);
         db.add_entity<GoalLink>(GoalLink.entity_mapping);
-        db.add_entity<GoalEvent>(GoalEvent.entity_mapping);
         db.migrate_on_startup();
 
         application.add_scoped<ProjectService>();

src/pages/home.html

@@ -239,7 +239,7 @@
                         <td><span stm-attribute.class="'badge badge-' + item.kind" stm-text="item.kind">kind</span>
                             <!-- Comment rows read as footnotes (plan 13
                                  B4): the body leads, muted and smaller. -->
-                            <span stm-attribute.class="item.kind == 'comment' || item.kind == 'goal-comment' ? 'footnote' : ''"
+                            <span stm-attribute.class="item.kind == 'comment' || item.kind == 'goal-comment' || item.kind == 'goal-system' ? 'footnote' : ''"
                                   stm-text="item.text">activity</span></td>
                     </tr>
                     <tr stm-if="dashboard.activity.length == 0"><td colspan="3" class="muted">Nothing yet.</td></tr>

test/integration_test.py

@@ -27,7 +27,7 @@ The goal-era arc (plan 12):
    completion report
 6. completion review accept -> goal complete, run task done
 7. the goal agent's terminal sessions (SSE replay), the audit trail
-   (goal_events narrative, typed session_commands), the /status
+   (the derived goal narrative, typed session_commands), the /status
    goal_branches cache sampled by the manual saddle refresh
 8. the legacy ingest API surface (401s, typed updates, exactly-once
    poll handover) and the stm-markdown renderer
@@ -526,10 +526,11 @@ def main():
         check(kept, "the ephemeral branch survives in the goal clone for the merge")
 
         print("== coalesced resume -> merge -> push -> completion review ==")
-        check(wait_for(lambda: db_query(
-            "select 1 from goal_events where goal_id = ? and kind = 'task-done'",
-            (goal1,)) or False, timeout=30, what="the task-done event") is not None,
-              "the worker's terminal event lands in the goal narrative")
+        check(wait_for(lambda: any(
+            e["kind"] == "task-done" and int(e["task_id"]) == worker_task
+            for e in goal_state(goal1).get("events", [])) or False,
+            timeout=30, what="the task-done narrative row") is not None,
+              "the worker's terminal transition lands in the derived goal narrative")
         # the goal agent's /status reports the branch push state while it
         # works; sample the moment the resumed agent is mid-round, then
         # cache it through the manual saddle refresh
@@ -562,20 +563,26 @@ def main():
         print("== completion accept ==")
         completion = state["completion_pending_block"]
         check(completion and completion.get("body", "") == state.get("completion_report", ""),
-              "the completion block mirrors the report")
+              "the run task's completion block carries the report")
         act(auth, completion["answer_action"], {"answer": REVIEW_APPROVED})
         check(wait_goal_status(auth, goal1, "complete", timeout=30) is not None,
               "accepting with the agreed wording completes the goal")
         check(wait_task_status(auth, run_task, "done", timeout=30) is not None,
               "the goal-run task closes done with the goal")
 
-        print("== audit trail ==")
-        kinds = [row[0] for row in db_query(
-            "select kind from goal_events where goal_id = ? order by id", (goal1,))]
+        print("== audit trail (the derived narrative) ==")
+        narrative = goal_state(goal1)["events"]
+        kinds = [e["kind"] for e in narrative]
         for expected in ("started", "block-raised", "plan-submitted", "plan-rejected",
-                         "plan-approved", "task-created", "task-enqueued", "task-done",
+                         "plan-approved", "task-created", "task-done",
                          "completion-submitted", "goal-approved"):
-            check(expected in kinds, "goal_events records %s" % expected)
+            check(expected in kinds, "the derived narrative records %s" % expected)
+        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,))}

test/web_stack_test.py

@@ -2058,8 +2058,8 @@ def bundle_pages(ctx):
           and re.search(r"\.comment\s*\{[^}]*font-size:\s*0\.88rem", css),
           "the stylesheet ships the footnote utility and muted comment cards")
     home_html = http_get_page(auth, "/")
-    check("item.kind == 'comment' || item.kind == 'goal-comment' ? 'footnote'"
-          in home_html,
+    check("item.kind == 'comment' || item.kind == 'goal-comment'"
+          " || item.kind == 'goal-system' ? 'footnote'" in home_html,
           "the dashboard timeline renders comment rows as footnotes")
 
 
@@ -2474,9 +2474,11 @@ def bundle_dispatch(ctx):
     gpu_task = task_id_tmp
     detail = task_state(auth, gpu_task)
     act(auth, detail["attach_capability"], {"capability_id": str(int(caps["gpu"]["id"]))})
-    detail = task_state(auth, gpu_task)
-    check([c["name"] for c in detail["capabilities"]] == ["gpu"],
+    check(wait_for(lambda: [c["name"] for c in
+                            task_state(auth, gpu_task)["capabilities"]] == ["gpu"]
+                   or False, timeout=30, what="capability attached") is not None,
           "task page lists the required capability")
+    detail = task_state(auth, gpu_task)
     # The only idle saddle lacks the tag, so the queue flags the wait
     # (read before any other task claims the idle saddle).
     queue = [t for t in slot_public(auth, "/tasks", "tasks")["tasks"]
@@ -2924,9 +2926,13 @@ def bundle_subtasks(ctx):
     fake.goal_script = [("question", "Which harness: pytest or vitest?",
                          ["pytest", "vitest"])]
     conn = db()
-    conn.execute("insert into goal_events (goal_id, kind, text, created_at)"
-                 " values (?, 'task-done', 'Task #999 Nudge — done',"
-                 " strftime('%s','now','-5 seconds'))", (goal,))
+    # Plan 15 C3: a terminal transition (a done task row newer than
+    # the watermark) is the resume signal — no event rows exist.
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, completed_at, created_at) values (1, ?, 'task',"
+                 " 'Nudge', 'done', 5,"
+                 " strftime('%s','now'),"
+                 " strftime('%s','now','-30 seconds'))", (goal,))
     conn.commit()
     conn.close()
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "blocked"
@@ -4539,7 +4545,7 @@ def bundle_answers(ctx):
     # The hold_state daemon drives its round manually: the resumed
     # round completes on the fresh session row.
     parked.push_results("done", summary="parked round resumed and finished")
-    check(wait_task_done(auth, parked_task, timeout=30) is not None,
+    check(wait_task_done(auth, parked_task, timeout=60) is not None,
           "the parked session continues to completion")
     conn = db()
     row = conn.execute("select outcome from sessions where id = ?",
@@ -5154,9 +5160,11 @@ def bundle_push(ctx):
     other_before2 = len(push.hits("/other"))
     main_before2 = len(push.hits("/main"))
     conn = db()
-    conn.execute("insert into goal_events (goal_id, kind, text, created_at)"
-                 " values (?, 'task-done', 'Task #999 Warm — done',"
-                 " strftime('%s','now','-5 seconds'))", (push_goal,))
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, completed_at, created_at) values (1, ?, 'task',"
+                 " 'Warm', 'done', 5,"
+                 " strftime('%s','now'),"
+                 " strftime('%s','now','-30 seconds'))", (push_goal,))
     conn.commit()
     conn.close()
     check(wait_for(lambda: goal_state(push_goal)["status"] == "failed" or False,
@@ -5604,11 +5612,8 @@ def bundle_goals(ctx):
     conn.execute("insert into tasks (project_id, goal_id, kind, title, status, priority,"
                  " summary, completed_at, created_at) values (1, ?, 'task',"
                  " 'Quiet task', 'done', 5, 'finished while paused',"
-                 " strftime('%s','now','+1 second'), strftime('%s','now'))", (goal1,))
+                 " strftime('%s','now'), strftime('%s','now'))", (goal1,))
     quiet_id = conn.execute("select max(id) from tasks").fetchone()[0]
-    conn.execute("insert into goal_events (goal_id, kind, text, task_id, created_at)"
-                 " values (?, 'task-done', 'Task #? Quiet task — done: finished while"
-                 " paused', ?, strftime('%s','now','+1 second'))", (goal1, quiet_id))
     conn.commit()
     conn.close()
     paused_runs = len(fake.goal_assignments)
@@ -5628,6 +5633,13 @@ def bundle_goals(ctx):
     check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
                    or False, timeout=60, what="goal agent parks again") is not None,
           "the goal agent parks again after the flush")
+    # The watermark consumed the flushed transition: with nothing new
+    # under the goal, no further resume fires (parked stays parked).
+    flushed_assignments = len(fake.goal_assignments)
+    check(quiet_window(quiet_seconds(),
+                       lambda: len(fake.goal_assignments) == flushed_assignments,
+                       "consumed watermark never re-fires"),
+          "the consumed watermark never re-fires a resume")
 
     print("== plan 12: completion review, push state, accept ==")
     # The daemon exposes goal-branch push state on /status; the web
@@ -5660,13 +5672,15 @@ def bundle_goals(ctx):
 
     # `saddle done` in goal context: the completion review.
     fake.goal_script = [("done", "# Completion\n\nThe flange ships.")]
-    # Wake the parked agent: a pending event has waited out the window
-    # (the pause-flush above consumed the queue, so arm one directly —
-    # the queued 'Quiet task' row was consumed; use a fresh event).
+    # Wake the parked agent: a terminal transition newer than the
+    # watermark (the pause-flush consumed everything up to it, so arm
+    # one directly with a fresh row).
     conn = db()
-    conn.execute("insert into goal_events (goal_id, kind, text, created_at)"
-                 " values (?, 'task-done', 'Task #999 Warmup — done',"
-                 " strftime('%s','now','-5 seconds'))", (goal1,))
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, completed_at, created_at) values (1, ?, 'task',"
+                 " 'Warmup', 'done', 5,"
+                 " strftime('%s','now'),"
+                 " strftime('%s','now','-30 seconds'))", (goal1,))
     conn.commit()
     conn.close()
     check(wait_for(lambda: goal_state(goal1)["status"] == "completion_review" or False,
@@ -5692,9 +5706,11 @@ def bundle_goals(ctx):
     # Round 2: submit again and accept.
     fake.goal_script = [("done", "# Completion v2\n\nThe flange ships, with docs.")]
     conn = db()
-    conn.execute("insert into goal_events (goal_id, kind, text, created_at)"
-                 " values (?, 'task-done', 'Task #998 Warmup2 — done',"
-                 " strftime('%s','now','-5 seconds'))", (goal1,))
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, completed_at, created_at) values (1, ?, 'task',"
+                 " 'Warmup2', 'done', 5,"
+                 " strftime('%s','now'),"
+                 " strftime('%s','now','-30 seconds'))", (goal1,))
     conn.commit()
     conn.close()
     check(wait_for(lambda: goal_state(goal1)["status"] == "completion_review" or False,
@@ -5714,6 +5730,30 @@ def bundle_goals(ctx):
                    timeout=30, what="goal run task closes") is not None,
           "approving the completion closes the goal-run task done")
 
+    print("== plan 15 C3: the narrative derives from the recorded shape ==")
+    # Content equivalence with the old projection on this run's exact
+    # shape: every kind the goal_events projection produced, keyed the
+    # same way, plus the watermark and no-ingest guarantees.
+    narrative = goal_state(goal1)["events"]
+    kinds = [e["kind"] for e in narrative]
+    for expected in ("created", "started", "plan-submitted", "plan-rejected",
+                     "plan-approved", "task-created", "task-done",
+                     "completion-submitted", "completion-rejected", "goal-approved"):
+        check(expected in kinds,
+              "the derived narrative carries the %s row" % expected)
+    check(all(k != "task-enqueued" and k != "resumed" and k != "retry" for k in kinds),
+          "timestamp-less bookkeeping kinds never return")
+    stamps = [e["at_display"] for e in narrative if e.get("at_display")]
+    check(stamps == sorted(stamps, reverse=True),
+          "the narrative reads newest first")
+    conn = db()
+    no_event_table = conn.execute(
+        "select count(*) from sqlite_master where type = 'table'"
+        " and name = 'goal_events'").fetchone()[0] == 0
+    conn.close()
+    check(no_event_table,
+          "the goal_events table is gone (no narrative rows on ingest)")
+
     print("== plan 12: cancel cascades to the goal's tasks ==")
     goal2 = create_goal({"project_id": "1", "title": "Doomed goal",
                          "brief": "This goal will be cancelled.",
@@ -6150,14 +6190,10 @@ def bundle_replay(ctx):
         return slot_public(auth, "/goals/%d" % goal_id, "goal")
 
     def goal_failed_events(goal_id):
-        conn = db()
-        try:
-            return conn.execute(
-                "select count(*) from goal_events"
-                " where goal_id = ? and kind = 'goal-failed'",
-                (goal_id,)).fetchone()[0]
-        finally:
-            conn.close()
+        # Plan 15 C3: the narrative derives at read time — the
+        # failed-goal row IS the goal-failed record (one row, ever).
+        return len([e for e in goal_state(goal_id)["events"]
+                    if e["kind"] == "goal-failed"])
 
     act(auth, slot_public(auth, "/goals", "goals")["create"], {
         "project_id": "1", "title": "Burst goal", "brief": "Fails honestly.",
@@ -6258,9 +6294,6 @@ def bundle_replay(ctx):
     conn = db()
     task_comment_bodies = [r[0] for r in conn.execute(
         "select body from task_comments where task_id = ?", (comment_run,))]
-    event_texts = [r[0] for r in conn.execute(
-        "select text from goal_events where goal_id = ? and kind = 'comment'",
-        (comment_goal,))]
     tables = {r[0] for r in conn.execute(
         "select name from sqlite_master where type = 'table'").fetchall()}
     conn.close()
@@ -6268,9 +6301,13 @@ def bundle_replay(ctx):
           "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(len(event_texts) == 1 and event_texts[0].startswith("The flange weld holds.")
-          and "saddle comment" not in event_texts[0] and "EOF" not in event_texts[0],
-          "the goal_events comment projection stores the body text")
+    narrative_comments = [e["text"] for e in goal_state(comment_goal)["events"]
+                          if e["kind"] == "comment"]
+    check(len(narrative_comments) == 1
+          and narrative_comments[0].startswith("The flange weld holds.")
+          and "saddle comment" not in narrative_comments[0]
+          and "EOF" not in narrative_comments[0],
+          "the derived narrative's comment rows carry the body text")
 
     state = goal_state(comment_goal)
     check(state["current_command"] == comment_body,
@@ -6571,7 +6608,8 @@ def bundle_goalui(ctx):
     act(auth, state["requeue_run"], {})
     check(task_state(auth, run_task)["status"] == "queued",
           "the requeue returns the run to the queue")
-    check(open_session(run_task) is None,
+    check(wait_for(lambda: open_session(run_task) is None or False,
+                   timeout=30, what="parked session closed") is not None,
           "the requeue closed the parked session (restore-on-dispatch)")
     conn = db()
     conn.execute("update saddles set status = 'idle' where id = ?", (fake_id,))
@@ -6610,8 +6648,9 @@ def bundle_goalui(ctx):
                    or False, timeout=60, what="retried supervision parks") is not None,
           "the retried supervision dispatches and parks again")
     state = goal_state(goal)
-    check(any(e["kind"] == "retry" for e in state["events"]),
-          "the narrative records the supervision retry")
+    check(any(e["kind"] == "system" and "Supervision retried" in e["text"]
+              for e in state["events"]),
+          "the narrative records the supervision retry (a system row)")
 
     print("== B5: order, effective rules, capability badges ==")
     goal_page = http_get_page(auth, "/goals/%d" % goal)
@@ -6755,9 +6794,11 @@ def bundle_goalui(ctx):
     # block, and a redelivered pair nothing more.
     fake.goal_script = [("done", "# Completion\n\nOne block only.")]
     conn = db()
-    conn.execute("insert into goal_events (goal_id, kind, text, created_at)"
-                 " values (?, 'task-done', 'Task #997 Warmup — done',"
-                 " strftime('%s','now','-5 seconds'))", (goal,))
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, completed_at, created_at) values (1, ?, 'task',"
+                 " 'Warmup', 'done', 5,"
+                 " strftime('%s','now'),"
+                 " strftime('%s','now','-30 seconds'))", (goal,))
     conn.commit()
     conn.close()
     check(wait_for(lambda: goal_state(goal)["status"] == "completion_review" or False,