M0003 event indexes, batched chunks ingestion, LLM usage surfacing, resume failure signatures

- M0003_EventIndexes (serial 3): idx_events_scope(scope_type, scope_id, id)
  and the partial idx_events_terminal(scope_id, created_at) WHERE
  kind='state_change' AND scope_type='task' — every EventService reader
  full-scanned the 47k-row production ledger; EXPLAIN now SEARCHes both.
- store_pushed_events inserts a whole pushed batch in ONE transaction via a
  prepared INSERT OR IGNORE (duplicates consumed in SQLite, stored count from
  connection.changes — no per-row re-query); delete_for_task is one bulk
  DELETE by scope.
- task and goal pages render an LLM usage stat line (calls, prompt/completion
  tokens, wall time) aggregated from usage events — the fleet burned 964M
  prompt tokens in six days invisibly.
- take_resume_notice appends a REPEATED FAILURE section when 2+ worker tasks
  failed with the same 120-char summary signature since the park, telling the
  goal agent to change approach or split scope instead of recreating clones
  (production: tasks 101-105).
- goalui assertions follow the plan-review engine into the served app.js
  (stale after the New nav/Account dropdown theming; no product bugs).

1182/1 across all 18 bundles (the one is the known resurrect dispatch-timing
flake — 37/0 isolated; parity 21/0 and origins 15/0 run after its abort).

meson.build

@@ -178,6 +178,7 @@ app_sources = files(
     'src/Data/Entities.vala',
     'src/Data/Migrations/M0001_Initial.vala',
     'src/Data/Migrations/M0002_BatchImprovements.vala',
+    'src/Data/Migrations/M0003_EventIndexes.vala',
     'src/Services/ProjectService.vala',
     'src/Services/TaskService.vala',
     'src/Services/GoalService.vala',

src/Data/Migrations/M0003_EventIndexes.vala

@@ -0,0 +1,105 @@
+using InvercargillSql.Migrations;
+using Sqlite;
+
+namespace ClankerSaddle.Data.Migrations {
+
+    /**
+     * P1 (web-db health plan): reader indexes for the event ledger.
+     *
+     * The production `events` table (47k rows, 36 MB at diagnosis) has
+     * no reader indexes — only the UNIQUE(session_id, seq) autoindex —
+     * so every EventService reader (page_for_task, newer_than,
+     * list_for_task/goal, the resume coalescer's terminal-transition
+     * walks) full-scans the ledger on each call.
+     *
+     * Two indexes cover the read paths:
+     *   - idx_events_scope(scope_type, scope_id, id) — every
+     *     scope-filtered, id-ordered reader (covering: the id ordering
+     *     rides the index tail, so pages and SSE cursor walks never
+     *     sort).
+     *   - idx_events_terminal(scope_id, created_at) WHERE
+     *     kind = 'state_change' AND scope_type = 'task' — the resume
+     *     machinery's terminal-transition scans, as a PARTIAL index so
+     *     it stays small (state_change rows are a sliver of the ledger
+     *     and ingestion never touches it).
+     *
+     * Index API note: the framework's structured path (t.index() /
+     * IndexBuilder) expresses neither IF NOT EXISTS nor a partial
+     * WHERE clause (CreateIndexOperation carries only name, columns
+     * and uniqueness). MigrationBuilder.execute_sql — the framework's
+     * documented raw-SQL operation, executed by the runner inside the
+     * migration transaction — expresses both, so both indexes are
+     * created through it exactly as designed, partial WHERE included.
+     *
+     * IDEMPOTENCE: same contract as M0002 — the runner invokes up()
+     * BEFORE opening the apply transaction (the builder only queues),
+     * so up() introspects sqlite_master through a separate read-only
+     * handle and queues only what the schema still needs. A fresh
+     * database (M0001 just made `events`) queues both; an upgraded
+     * serial-2 database queues both; a database that already has them
+     * (defensive: this build's DB handed back to an older serial, then
+     * forward again) queues nothing and records as a no-op. The SQL
+     * itself carries IF NOT EXISTS as a second guard.
+     */
+    public class M0003_EventIndexes : Migration {
+
+        public override string name { get { return "Event ledger reader indexes"; } }
+        public override uint64 serial { get { return 3; } }
+        public override string migration_namespace { get { return "clanker-saddle"; } }
+
+        // The database this migration is upgrading — same resolution
+        // as M0002 (the runner never hands up() a connection, so the
+        // schema is read through a separate read-only handle on the
+        // same SPRY_DB_PATH main.vala connects).
+        private string db_path() {
+            var path = Environment.get_variable("SPRY_DB_PATH") ?? "";
+            if (path.strip().length == 0) {
+                return "./clanker-saddle.sqlite";
+            }
+            return path;
+        }
+
+        private bool index_exists(string name) {
+            Database db;
+            int rc = Database.open_v2(db_path(), out db, Sqlite.OPEN_READONLY);
+            if (rc != Sqlite.OK) {
+                // Unopenable (locked, or missing — a database the
+                // runner is about to create): treat as absent and let
+                // the queued operations fail loudly rather than
+                // silently skipping an upgrade.
+                return false;
+            }
+            Statement statement;
+            rc = db.prepare_v2(
+                "SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' AND name = ?",
+                -1, out statement);
+            if (rc != Sqlite.OK) {
+                return false;
+            }
+            statement.bind_text(1, name, -1);
+            if (statement.step() != Sqlite.ROW) {
+                return false;
+            }
+            return statement.column_int64(0) > 0;
+        }
+
+        public override void up(MigrationBuilder b) {
+            if (!index_exists("idx_events_scope")) {
+                b.execute_sql(
+                    "CREATE INDEX IF NOT EXISTS idx_events_scope"
+                    + " ON events(scope_type, scope_id, id)");
+            }
+            if (!index_exists("idx_events_terminal")) {
+                b.execute_sql(
+                    "CREATE INDEX IF NOT EXISTS idx_events_terminal"
+                    + " ON events(scope_id, created_at)"
+                    + " WHERE kind = 'state_change' AND scope_type = 'task'");
+            }
+        }
+
+        public override void down(MigrationBuilder b) {
+            b.execute_sql("DROP INDEX IF EXISTS idx_events_terminal");
+            b.execute_sql("DROP INDEX IF EXISTS idx_events_scope");
+        }
+    }
+}

src/Format.vala

@@ -1,3 +1,5 @@
+using ClankerSaddle.Services;
+
 namespace ClankerSaddle {
 
     /**
@@ -11,4 +13,74 @@ namespace ClankerSaddle {
     public string format_stamp(DateTime? stamp) {
         return stamp != null ? ((!)stamp).format("%Y-%m-%d %H:%M") : "";
     }
+
+    /** A decimal integer grouped with commas (1234567 -> "1,234,567"). */
+    public string format_grouped(int64 value) {
+        string raw = value.to_string();
+        var builder = new StringBuilder();
+        for (int at = 0; at < raw.length; at++) {
+            builder.append_c(raw[at]);
+            int remaining = raw.length - at;
+            if (remaining > 3 && remaining % 3 == 1) {
+                builder.append_c(',');
+            }
+        }
+        return builder.str;
+    }
+
+    /**
+     * Token magnitudes for the usage stat line (S5): millions and
+     * tens-of-thousands read as one-decimal magnitudes ("45.2M",
+     * "84.1K"); anything smaller stays exact and comma-grouped —
+     * the ranges where operators compare calls byte-for-byte.
+     */
+    public string format_tokens(int64 tokens) {
+        if (tokens >= 1000000) {
+            return "%.1fM".printf(tokens / 1000000.0);
+        }
+        if (tokens >= 10000) {
+            return "%.1fK".printf(tokens / 1000.0);
+        }
+        return format_grouped(tokens);
+    }
+
+    /**
+     * Wall-time durations for the usage stat line (S5): the two
+     * leading non-zero units — "2d 3h", "3h 12m", "4m 09s" style,
+     * bare seconds under a minute.
+     */
+    public string format_duration_ms(int64 elapsed_ms) {
+        int64 total_seconds = elapsed_ms / 1000;
+        int days = (int) (total_seconds / 86400);
+        int hours = (int) ((total_seconds % 86400) / 3600);
+        int minutes = (int) ((total_seconds % 3600) / 60);
+        int seconds = (int) (total_seconds % 60);
+        if (days > 0) {
+            return "%dd %dh".printf(days, hours);
+        }
+        if (hours > 0) {
+            return "%dh %dm".printf(hours, minutes);
+        }
+        if (minutes > 0) {
+            return "%dm %ds".printf(minutes, seconds);
+        }
+        return "%ds".printf(seconds);
+    }
+
+    /**
+     * The compact LLM usage stat line (S5) for the task and goal
+     * pages — "LLM: 1,234 calls · 45.2M in / 1.1M out tok ·
+     * 3h 12m". An empty summary formats as "" (the pages render
+     * nothing for zero usage).
+     */
+    public string format_usage_line(UsageSummary usage) {
+        if (usage.is_empty()) {
+            return "";
+        }
+        return "LLM: %s calls · %s in / %s out tok · %s".printf(
+            format_grouped(usage.calls),
+            format_tokens(usage.prompt_tokens),
+            format_tokens(usage.completion_tokens),
+            format_duration_ms(usage.elapsed_ms));
+    }
 }

src/Services/EventService.vala

@@ -1,6 +1,7 @@
 using Invercargill;
 using Invercargill.DataStructures;
 using Invercargill.Expressions;
+using InvercargillSql;
 using InvercargillSql.Orm;
 using Inversion;
 using ClankerSaddle.Data;
@@ -38,6 +39,24 @@ namespace ClankerSaddle.Services {
     /** The event log page/SSE window (plan 17 §3.2). */
     public const int EVENT_PAGE_SIZE = 30;
 
+    /**
+     * An aggregate LLM-usage roll-up (S5): the COUNT of `usage`
+     * events under one scope plus the SUMs of their payload's
+     * prompt/completion tokens and per-call elapsed time — the
+     * source of the task and goal pages' usage stat line.
+     */
+    public class UsageSummary : Object {
+        public int64 calls { get; set; }
+        public int64 prompt_tokens { get; set; }
+        public int64 completion_tokens { get; set; }
+        public int64 elapsed_ms { get; set; }
+
+        /** True when the scope recorded no usage events at all. */
+        public bool is_empty() {
+            return calls <= 0;
+        }
+    }
+
     /**
      * The plan-17 unified event ledger's writer and read model: daemon
      * event ingestion from `POST /api/saddle/chunks` (idempotent on
@@ -48,6 +67,7 @@ namespace ClankerSaddle.Services {
     public class EventService : Object {
 
         private OrmSession db = inject<OrmSession>();
+        private Connection connection = inject<Connection>();
 
         // ------------------------------------------------------------------
         // Writers
@@ -82,10 +102,19 @@ namespace ClankerSaddle.Services {
          * daemon contract makes seq mandatory). The daemon's `ts` is
          * authoritative for `created_at` (fallback: now). Returns the
          * count of rows newly stored.
+         *
+         * P2 (web-db health plan): the whole batch inserts inside ONE
+         * transaction through one prepared INSERT OR IGNORE statement
+         * — SQLite consumes a duplicate (session_id, seq) inside the
+         * statement itself (no per-row conflict round trip, no
+         * per-row re-query), and the batch's accumulated changes
+         * count is exactly the newly stored rows (an ignored row
+         * contributes 0). Staging stays as before: validation runs to
+         * completion first, so one invalid event (bad seq/kind)
+         * refuses the whole push with nothing stored.
          */
         public async int64 store_pushed_events(int64 session_id, int64 task_id,
                 Json.Array events) throws Error {
-            int64 stored = 0;
             var staged = new Vector<EventRecord>();
             foreach (var element in events.get_elements()) {
                 if (element.get_node_type() != Json.NodeType.OBJECT) {
@@ -121,19 +150,24 @@ namespace ClankerSaddle.Services {
                         ?? new DateTime.now_local()
                 });
             }
+            var mapper = db.get_mapper<EventRecord>();
+            var parameter_sets = new Vector<Properties>();
             foreach (var record in staged) {
-                try {
-                    yield db.insert_async<EventRecord>(record);
-                    stored++;
-                } catch (Error insert_error) {
-                    var existing = yield db.query<EventRecord>()
-                        .where(expr("session_id == $0 && seq == $1",
-                            elem<int64?>(session_id), elem<int64?>(record.seq)))
-                        .first_async();
-                    if (existing == null) {
-                        throw insert_error;
-                    }
-                }
+                parameter_sets.add(mapper.map_from(record));
+            }
+            var transaction = connection.begin_transaction();
+            int64 stored = 0;
+            try {
+                stored = connection.create_command(
+                    "INSERT OR IGNORE INTO events"
+                    + " (session_id, seq, kind, actor, scope_type, scope_id, payload, created_at)"
+                    + " VALUES (:session_id, :seq, :kind, :actor, :scope_type, :scope_id,"
+                    + " :payload, :created_at)")
+                    .execute_batch(parameter_sets);
+                transaction.commit();
+            } catch (Error batch_error) {
+                transaction.rollback();
+                throw batch_error;
             }
             return stored;
         }
@@ -305,14 +339,84 @@ namespace ClankerSaddle.Services {
             return newest;
         }
 
-        /** Deletes every event scoped to the task (the delete cascade). */
-        public async void delete_for_task(int64 task_id) throws Error {
-            foreach (var record in yield db.query<EventRecord>()
-                    .where(expr("scope_type == $0 && scope_id == $1",
-                        elem(EVENT_SCOPE_TASK), elem<int64?>(task_id)))
-                    .materialise_async()) {
-                yield db.delete_async<EventRecord>(record);
+        /**
+         * The task's aggregate LLM usage (S5): COUNT and SUMs over
+         * the `usage` events scoped to the task, the payload read
+         * with json_extract — the daemon's prompt_tokens/
+         * completion_tokens/elapsed_ms shape, with the pre-rename
+         * input_tokens/output_tokens keys as a fallback so older
+         * ledgers still roll up. No usage events aggregate to an
+         * empty summary.
+         */
+        public async UsageSummary usage_summary(int64 task_id) throws Error {
+            var command = connection.create_command(usage_summary_sql(
+                "scope_type = :scope AND scope_id = :task"));
+            command.with_parameter<Invercargill.Element>("kind", elem(EVENT_USAGE));
+            command.with_parameter<Invercargill.Element>("scope", elem(EVENT_SCOPE_TASK));
+            command.with_parameter<Invercargill.Element>("task", elem<int64?>(task_id));
+            return yield materialise_usage(command);
+        }
+
+        /**
+         * The goal's aggregate LLM usage (S5): the same roll-up
+         * summed across EVERY task of the goal (the agent's own run
+         * plus its workers — keyed on tasks.goal_id; the run task's
+         * own page redirects to the goal, so this line is the run's
+         * only cost surface).
+         */
+        public async UsageSummary usage_summary_for_goal(int64 goal_id) throws Error {
+            var command = connection.create_command(usage_summary_sql(
+                "scope_type = :scope AND scope_id IN"
+                + " (SELECT id FROM tasks WHERE goal_id = :goal)"));
+            command.with_parameter<Invercargill.Element>("kind", elem(EVENT_USAGE));
+            command.with_parameter<Invercargill.Element>("scope", elem(EVENT_SCOPE_TASK));
+            command.with_parameter<Invercargill.Element>("goal", elem<int64?>(goal_id));
+            return yield materialise_usage(command);
+        }
+
+        /** The shared aggregate SELECT over the usage payload. */
+        private static string usage_summary_sql(string scope_filter) {
+            return "SELECT COUNT(*) AS calls"
+                + ", SUM(COALESCE(CAST(json_extract(payload, '$.prompt_tokens') AS INTEGER)"
+                + ", CAST(json_extract(payload, '$.input_tokens') AS INTEGER), 0)) AS prompt_tokens"
+                + ", SUM(COALESCE(CAST(json_extract(payload, '$.completion_tokens') AS INTEGER)"
+                + ", CAST(json_extract(payload, '$.output_tokens') AS INTEGER), 0)) AS completion_tokens"
+                + ", SUM(CAST(json_extract(payload, '$.elapsed_ms') AS INTEGER)) AS elapsed_ms"
+                + " FROM events WHERE kind = :kind AND " + scope_filter;
+        }
+
+        /** Runs the aggregate command; exactly one row (NULL sums read 0). */
+        private async UsageSummary materialise_usage(Command command) throws Error {
+            var summary = new UsageSummary();
+            foreach (var row in yield command.execute_query_async()) {
+                summary.calls = usage_number(row.get("calls"));
+                summary.prompt_tokens = usage_number(row.get("prompt_tokens"));
+                summary.completion_tokens = usage_number(row.get("completion_tokens"));
+                summary.elapsed_ms = usage_number(row.get("elapsed_ms"));
             }
+            return summary;
+        }
+
+        /** One aggregate column back as an int64 (0 when null/unreadable). */
+        private static int64 usage_number(Invercargill.Element? element) {
+            if (element == null) {
+                return 0;
+            }
+            int64? value = 0;
+            return ((!)element).try_get_as<int64?>(out value) ? (value ?? 0) : 0;
+        }
+
+        /**
+         * Deletes every event scoped to the task (the delete cascade).
+         * One bulk DELETE — the scope index serves the predicate; no
+         * materialise-then-delete-per-row round trips (P2).
+         */
+        public async void delete_for_task(int64 task_id) throws Error {
+            yield connection.create_command(
+                "DELETE FROM events WHERE scope_type = :scope_type AND scope_id = :scope_id")
+                .with_parameter<Invercargill.Element>("scope_type", elem(EVENT_SCOPE_TASK))
+                .with_parameter<Invercargill.Element>("scope_id", elem<int64?>(task_id))
+                .execute_non_query_async();
         }
 
         // ------------------------------------------------------------------

src/Services/GoalService.vala

@@ -1294,6 +1294,7 @@ namespace ClankerSaddle.Services {
             report.append("Goal update — resuming your parked session:\n\n");
             report.append("## Task statuses\n");
             var finished = new Vector<TaskEntity>();
+            var new_failures = new Vector<TaskEntity>();
             foreach (var task in yield db.query<TaskEntity>()
                     .where(expr("goal_id == $0 && kind == $1",
                         elem<int64?>(goal_id), elem(TASK_KIND_TASK)))
@@ -1309,6 +1310,13 @@ namespace ClankerSaddle.Services {
                 } else if (task.status == STATUS_FAILED) {
                     report.append_printf("- #%s %s — failed: %s\n", task.id.to_string(),
                         task.title, notice_lead(task));
+                    // S6: the failures newer than the park feed the
+                    // repeated-signature clustering below (the same
+                    // park-relative rule the new completions use).
+                    if (since == null || task.completed_at == null
+                            || ((!)task.completed_at).compare((!)since) > 0) {
+                        new_failures.add(task);
+                    }
                 } else if (task.status == STATUS_DRAFT) {
                     // Goal 4 task 1: the goal-2 stall class — a draft
                     // that was never enqueued produces no transitions,
@@ -1326,6 +1334,16 @@ namespace ClankerSaddle.Services {
                         task.summary ?? "no summary reported");
                 }
             }
+            // S6: repeated-failure signature clusters — worker tasks
+            // that failed since the park with IDENTICAL summary
+            // signatures (the fail-loop shape: a re-cloned task
+            // hitting the same deterministic wall). The section
+            // tells the agent to change its approach instead of
+            // recreating the same task again.
+            string repeated_failures = repeated_failure_section(new_failures);
+            if (repeated_failures.length > 0) {
+                report.append(repeated_failures);
+            }
             report.append("\nSupervise your tasks: merge completed work into the goal branch,")
                 .append(" prune worktrees, create or enqueue follow-ups as needed, then")
                 .append(" `saddle suspend` again when you are parked.\n");
@@ -1375,6 +1393,72 @@ namespace ClankerSaddle.Services {
             return lead.strip();
         }
 
+        /**
+         * The S6 failure signature: the trimmed summary's first ~120
+         * characters — long enough to capture a provider error's
+         * distinctive head (byte-identical fail-loop summaries match
+         * exactly), short enough to survive trailing per-task noise.
+         */
+        private const int FAILURE_SIGNATURE_CHARS = 120;
+
+        /** The signature of a failed task's summary ("" when none). */
+        private static string failure_signature(TaskEntity task) {
+            string summary = (task.summary ?? "").strip();
+            return summary.length > FAILURE_SIGNATURE_CHARS
+                ? summary.substring(0, FAILURE_SIGNATURE_CHARS) : summary;
+        }
+
+        /**
+         * The repeated-failure section of a resume notice (S6):
+         * groups the (park-relative) failed worker tasks by summary
+         * signature; any group of two or more renders the loud
+         * change-your-approach instruction with the member ids and
+         * the signature's first line. "" when no cluster exists —
+         * single failures stay ordinary status lines.
+         */
+        private static string repeated_failure_section(Vector<TaskEntity> failures) {
+            var order = new Vector<string>();
+            var groups = new HashTable<string, Vector<TaskEntity>>(str_hash, str_equal);
+            foreach (var task in failures) {
+                string signature = failure_signature(task);
+                if (signature.length == 0) {
+                    continue;
+                }
+                var group = groups.get(signature);
+                if (group == null) {
+                    group = new Vector<TaskEntity>();
+                    groups.set(signature, (!)group);
+                    order.add(signature);
+                }
+                ((!)group).add(task);
+            }
+            var section = new StringBuilder();
+            foreach (string signature in order) {
+                var group = groups.get(signature) ?? new Vector<TaskEntity>();
+                if (((!)group).length < 2) {
+                    continue;
+                }
+                if (section.len == 0) {
+                    section.append("\n## Repeated failure signature\n\n");
+                }
+                var ids = new StringBuilder();
+                foreach (var task in (!)group) {
+                    if (ids.len > 0) {
+                        ids.append(", ");
+                    }
+                    ids.append_printf("#%s", task.id.to_string());
+                }
+                int cut = signature.index_of("\n");
+                string lead = (cut >= 0 ? signature.substring(0, cut) : signature).strip();
+                section.append_printf(
+                    "REPEATED FAILURE: %d tasks failed with the same error signature —"
+                    + " do NOT recreate the same task again; change the approach, narrow"
+                    + " the mandate, or split the scope.\n- %s — shared signature: %s\n",
+                    (int) ((!)group).length, ids.str, lead);
+            }
+            return section.str;
+        }
+
         // ------------------------------------------------------------------
         // Completion gate (plan 15 C4)
         // ------------------------------------------------------------------

src/actions/TaskActions.vala

@@ -66,7 +66,7 @@ namespace ClankerSaddle {
             var me = SpryAuth.current(request);
             return yield TaskDetailState.build(state_service, task_service,
                 project_service, saddle_service, rule_service, capability_service,
-                action_registry, poll_state, push_service, db,
+                action_registry, poll_state, push_service, event_service, db,
                 me != null ? ((!)me).user_id : 0, task_id);
         }
     }

src/entrypoints/GoalDetailEntrypoint.vala

@@ -184,6 +184,12 @@ namespace ClankerSaddle {
          * for the parked ones ("" when there is nothing to say).
          */
         public string current_activity { get; set; default = ""; }
+        /**
+         * The goal's aggregate LLM usage stat line (S5), summed
+         * across every task of the goal (the agent's run plus its
+         * workers) — "" renders nothing.
+         */
+        public string usage_line { get; set; default = ""; }
         public Statum.Model.ActionDto? start { get; set; }
         /**
          * The draft brief-card editor's save (title/brief/priority):
@@ -564,6 +570,12 @@ namespace ClankerSaddle {
                 pub.compare_url = @"$(TaskService.https_git_url(((!)project).git_url))/compare/$(base_name)...$(goal.branch_name)";
             }
 
+            // S5: the goal's aggregate usage across all its tasks
+            // ("" when none recorded — the status head renders no
+            // line then).
+            pub.usage_line = format_usage_line(
+                yield ctx.events.usage_summary_for_goal(goal.id));
+
             if (ctx.actions != null) {
                 var sealed = new GoalActionPrivate() { goal_id = goal.id };
                 pub.comment = ((!)ctx.actions).author_private<GoalCommentAction, GoalActionPrivate>(sealed);

src/entrypoints/TaskDetailEntrypoint.vala

@@ -111,6 +111,12 @@ namespace ClankerSaddle {
          */
         public Statum.Model.ActionDto? force_wake { get; set; }
         public string saddle_health { get; set; default = ""; }
+        /**
+         * The aggregate LLM usage stat line (S5) — "LLM: 1,234
+         * calls · 45.2M in / 1.1M out tok · 3h 12m"; "" when the
+         * task recorded no usage events (the page renders nothing).
+         */
+        public string usage_line { get; set; default = ""; }
         public string display { get; set; default = ""; }
         public bool terminal { get; set; }
         public bool failed { get; set; }
@@ -177,7 +183,7 @@ namespace ClankerSaddle {
         public static async Slot build(StateService state_service, TaskService tasks,
                 ProjectService projects, SaddleService saddles, RuleService rules,
                 CapabilityService capabilities, ActionRegistry actions,
-                SaddlePollState polls, PushService push,
+                SaddlePollState polls, PushService push, EventService events,
                 InvercargillSql.Orm.OrmSession db, int64 viewer_id,
                 int64 task_id) throws GLib.Error {
             var detail = yield tasks.get_task(task_id);
@@ -189,6 +195,11 @@ namespace ClankerSaddle {
             var task = ((!)detail).task;
             var project = yield projects.get_project(task.project_id);
 
+            // S5: the aggregate usage roll-up ("" when no usage
+            // events — the status card renders no line then).
+            string usage_line = format_usage_line(
+                yield events.usage_summary(task.id));
+
             var pub = new TaskDetailPublic() {
                 exists = true,
                 id = task.id,
@@ -209,7 +220,8 @@ namespace ClankerSaddle {
                 enqueue_after_task_id = task.enqueue_after_task_id,
                 created_display = format_stamp(task.created_at),
                 started_display = format_stamp(task.started_at),
-                completed_display = format_stamp(task.completed_at)
+                completed_display = format_stamp(task.completed_at),
+                usage_line = usage_line
             };
 
             var display = TaskService.display_of_detail(((!)detail));
@@ -509,6 +521,7 @@ namespace ClankerSaddle {
         private CapabilityService capability_service = Inversion.inject<CapabilityService>();
         private SaddlePollState poll_state = Inversion.inject<SaddlePollState>();
         private PushService push_service = Inversion.inject<PushService>();
+        private EventService event_service = Inversion.inject<EventService>();
         private InvercargillSql.Orm.OrmSession db = Inversion.inject<InvercargillSql.Orm.OrmSession>();
         private UserService users = Inversion.inject<UserService>();
 
@@ -522,7 +535,7 @@ namespace ClankerSaddle {
             var me = SpryAuth.current(request);
             var state = yield TaskDetailState.build(state_service, task_service,
                 project_service, saddle_service, rule_service, capability_service,
-                action_registry, poll_state, push_service, db,
+                action_registry, poll_state, push_service, event_service, db,
                 me != null ? ((!)me).user_id : 0, task_id);
             // Plan 14 B1: the goal page fully replaces the goal-run
             // task's page — deep links (dashboard, notifications,

src/main.vala

@@ -44,6 +44,7 @@ int main(string[] args) {
         }
         db.add_migration<ClankerSaddle.Data.Migrations.M0001_Initial>();
         db.add_migration<ClankerSaddle.Data.Migrations.M0002_BatchImprovements>();
+        db.add_migration<ClankerSaddle.Data.Migrations.M0003_EventIndexes>();
         db.add_entity<Project>(Project.entity_mapping);
         db.add_entity<ProjectLink>(ProjectLink.entity_mapping);
         db.add_entity<TaskEntity>(TaskEntity.entity_mapping);

src/pages/goal_detail.html

@@ -189,6 +189,10 @@
                    stm-text="goal.current_activity">activity</p>
                 <p class="muted" stm-if="goal.saddle_health.length > 0"
                    stm-text="goal.saddle_health">health</p>
+                <!-- S5: the goal's aggregate LLM usage across its tasks
+                     (the run plus the workers) — empty renders nothing. -->
+                <p class="muted" stm-if="goal.usage_line.length > 0"
+                   stm-text="goal.usage_line">LLM: 1,234 calls · 45.2M in / 1.1M out tok · 3h 12m</p>
             </section>
 
             <!-- The agent's live progress feed (task-page pattern):

src/pages/task_detail.html

@@ -103,6 +103,10 @@
                     <p class="muted">Created <span stm-text="task.created_display">when</span></p>
                     <p class="muted" stm-if="task.started_display.length > 0">Started <span stm-text="task.started_display">when</span></p>
                     <p class="muted" stm-if="task.completed_display.length > 0">Completed <span stm-text="task.completed_display">when</span></p>
+                    <!-- S5: the aggregate LLM usage stat line (empty for
+                         zero-usage tasks renders nothing). -->
+                    <p class="muted" stm-if="task.usage_line.length > 0"
+                       stm-text="task.usage_line">LLM: 1,234 calls · 45.2M in / 1.1M out tok · 3h 12m</p>
                     <p stm-if="task.task_branch.length > 0">Branch: <strong stm-text="task.task_branch">branch</strong>
                         <span class="muted" stm-if="task.branch_from.length > 0"> from <span stm-text="task.branch_from">base</span></span></p>
                     <p stm-if="task.status == 'suspended'" class="muted">Parked (`saddle suspend`) —

test/bundles/goals.py

@@ -149,7 +149,7 @@ def run(ctx):
     # list page is iconed, and the dashboard pile's goal cards carry
     # their own accent styling.
     home_html = http_get_page(auth, "/")
-    check('<a href="/goals">Goals</a>' in home_html,
+    check('<a class="nav-goal" href="/goals">' in home_html,
           "the nav carries the Goals entry beside Tasks")
     check(any(p["item_type"] == "goal" for p in
               slot_public(auth, "/", "dashboard")["pinned"]),
@@ -514,6 +514,130 @@ def run(ctx):
                        "consumed watermark never re-fires"),
           "the consumed watermark never re-fires a resume")
 
+    print("== S6: repeated-failure signature in the resume notice ==")
+    # The fail-loop shape (web-db health plan A1/A2): worker tasks
+    # failing with IDENTICAL summary signatures since the park. The
+    # notice must name the cluster and forbid re-cloning; distinct
+    # signatures, or failures older than the park, must NOT cluster.
+    check(wait_for(run_park_second_passed, timeout=30,
+                   what="S6 park second passes") is not None,
+          "the S6 park stamp is in a strictly older second")
+    shared_sig = ("curl exited 22: JSON parse error: Invalid UTF-8 middle byte"
+                  " 0x5c — the clamped plan slice cuts a multi-byte character"
+                  " mid-character at byte 3000 of the shared goal plan")
+    conn = db()
+    for tail in (" (clone attempt one)", " (clone attempt two)"):
+        conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                     " priority, summary, completed_at, created_at) values"
+                     " (1, ?, 'task', 'Flange clone', 'failed', 5, ?,"
+                     " strftime('%s','now'), strftime('%s','now','-30 seconds'))",
+                     (goal1, shared_sig + tail))
+    clone_ids = [int(r[0]) for r in conn.execute(
+        "select id from tasks where title = 'Flange clone'").fetchall()]
+    conn.commit()
+    conn.close()
+    check(len(clone_ids) == 2,
+          "the S6 fixture lands two same-signature failed workers")
+    s6_before = len(fake.goal_assignments)
+    check(wait_for(lambda: len(fake.goal_assignments) > s6_before or False,
+                   timeout=60, what="S6 repeated-failure resume fires") is not None,
+          "the repeated failures resume the parked agent")
+    s6_notice = fake.goal_assignments[-1].get("resume_notice") or ""
+    check("REPEATED FAILURE" in s6_notice
+          and "do NOT recreate the same task again" in s6_notice
+          and "change the approach, narrow the mandate, or split the scope"
+          in s6_notice,
+          "the notice carries the repeated-failure instruction")
+    check(("#%d" % clone_ids[0]) in s6_notice
+          and ("#%d" % clone_ids[1]) in s6_notice,
+          "the repeated-failure section names both cluster members")
+    check("Invalid UTF-8 middle byte 0x5c" in s6_notice,
+          "the section names the shared signature's first line")
+    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
+                   or False, timeout=60, what="S6 agent re-parks") is not None,
+          "the S6 agent re-parks")
+    # Negative round: one pre-park failure sharing the signature (the
+    # watermark excludes it) plus one NEW failure with a distinct
+    # signature — neither forms a ≥2 cluster, so no section renders.
+    check(wait_for(run_park_second_passed, timeout=30,
+                   what="S6 second park second passes") is not None,
+          "the S6 negative-round park stamp is strictly older")
+    conn = db()
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, summary, completed_at, created_at) values"
+                 " (1, ?, 'task', 'Stale signature clone', 'failed', 5, ?,"
+                 " strftime('%s','now','-120 seconds'),"
+                 " strftime('%s','now','-150 seconds'))",
+                 (goal1, shared_sig + " (stale pre-park clone)"))
+    stale_id = conn.execute(
+        "select id from tasks where title = 'Stale signature clone'"
+    ).fetchone()[0]
+    conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                 " priority, summary, completed_at, created_at) values"
+                 " (1, ?, 'task', 'Distinct failure', 'failed', 5,"
+                 " 'tests failed: the flange widget lacks a widget prop',"
+                 " strftime('%s','now'), strftime('%s','now','-30 seconds'))",
+                 (goal1,))
+    distinct_id = conn.execute(
+        "select id from tasks where title = 'Distinct failure'"
+    ).fetchone()[0]
+    conn.commit()
+    conn.close()
+    s6_more = len(fake.goal_assignments)
+    check(wait_for(lambda: len(fake.goal_assignments) > s6_more or False,
+                   timeout=60, what="S6 negative-round resume fires") is not None,
+          "the negative-round failures resume the agent")
+    s6_next = fake.goal_assignments[-1].get("resume_notice") or ""
+    check("REPEATED FAILURE" not in s6_next,
+          "no repeated-failure section without a same-signature pair"
+          " newer than the park")
+    check(("#%d" % distinct_id) in s6_next and ("#%d" % stale_id) in s6_next,
+          "the negative-round notice still reports both failures as statuses")
+    check(wait_for(lambda: task_state(auth, run_task)["status"] == "suspended"
+                   or False, timeout=60, what="S6 negative agent re-parks")
+          is not None,
+          "the S6 negative round re-parks")
+
+    print("== S5: the goal page's aggregate LLM usage stat line ==")
+    # usage_summary_for_goal sums the usage events across every task
+    # of the goal (the agent's run plus its workers); a goal with no
+    # usage renders no line.
+    conn = db()
+    conn.execute("insert into goals (project_id, title, brief, branch_name,"
+                 " status, created_at) values (1, 'Usage goal', 'Sum the usage.',"
+                 " 'goal/usage', 'active', strftime('%s','now'))")
+    usage_goal = conn.execute(
+        "select id from goals where title = 'Usage goal'").fetchone()[0]
+    for title, payloads in (
+            ("Usage worker one", ({"prompt_tokens": 1000000, "completion_tokens": 5000,
+                                   "elapsed_ms": 3600000},
+                                  {"prompt_tokens": 500000, "completion_tokens": 2500,
+                                   "elapsed_ms": 1800000})),
+            ("Usage worker two", ({"prompt_tokens": 2500000, "completion_tokens": 12000,
+                                   "elapsed_ms": 5400000},))):
+        conn.execute("insert into tasks (project_id, goal_id, kind, title, status,"
+                     " priority, created_at) values (1, ?, 'task', ?, 'done', 5,"
+                     " strftime('%s','now'))", (usage_goal, title))
+        worker = conn.execute(
+            "select id from tasks where title = ?", (title,)).fetchone()[0]
+        for payload in payloads:
+            conn.execute(
+                "insert into events (kind, actor, scope_type, scope_id, payload,"
+                " created_at) values ('usage', 'harness', 'task', ?, ?,"
+                " strftime('%s','now'))",
+                (worker, json.dumps(payload)))
+    conn.commit()
+    conn.close()
+    state = goal_state(usage_goal)
+    check(state.get("usage_line") == "LLM: 3 calls · 4.0M in / 19.5K out tok · 3h 0m",
+          "the goal state sums usage across its tasks (%r)" % state.get("usage_line"))
+    goal_page = http_get_page(auth, "/goals/%d" % usage_goal)
+    check('stm-if="goal.usage_line.length &gt; 0"' in goal_page
+          and 'stm-text="goal.usage_line"' in goal_page,
+          "the goal page renders the usage stat line in the status head")
+    check(goal_state(goal_elsewhere).get("usage_line") == "",
+          "a goal whose tasks recorded no usage carries no stat line")
+
     print("== plan 12: completion review, push state, accept ==")
     # The daemon exposes goal-branch push state on /status; the web
     # caches it on the goal when it probes the goal agent's saddle.

test/bundles/goalui.py

@@ -132,12 +132,25 @@ def run(ctx):
     remarks_at = goal_page.index("data-plan-remarks")
     check(body_at < remarks_at,
           "the remarks queue renders at the bottom of the plan body")
-    for marker in ("data-plan-accept", "data-plan-toggle", "data-plan-submit",
-                   "data-plan-remove", "plan-popover", "remarksByBlock"):
+    for marker in ("data-plan-accept", "data-plan-toggle", "data-plan-submit"):
         check(marker in goal_page, "the remarks queue ships %s" % marker)
-    check("(index + 1) + '. '" in goal_page,
+    # The queue ENGINE moved out of the inline page script into the
+    # shared app.js (main.html's <script pstm-res-src="app.js">, the
+    # theming commits): the page keeps the review markup and the
+    # wiring, the engine keeps the machinery. The engine is asserted
+    # on the SERVED bundle the page loads (the pages.py statum.js
+    # pattern); main.html ships it ahead of the page's own script,
+    # so the wiring below cannot run before the engine exists.
+    status, app_js = http("GET", "/_statum/resource/app.js", timeout=15)
+    check(status == 200 and "window.initPlanReviews" in app_js,
+          "the served app.js ships the shared initPlanReviews engine")
+    check("window.initPlanReviews" in goal_page,
+          "the goal page references the shared engine (its review rides it)")
+    for marker in ("data-plan-remove", "plan-popover", "remarksByBlock"):
+        check(marker in app_js, "the remarks queue ships %s" % marker)
+    check("(index + 1) + '. '" in app_js,
           "the rejection payload numbers every queued remark")
-    check("rejectOpenByBlock" in goal_page and "statum.afterEvaluate(syncPlanRemarks)" in goal_page,
+    check("rejectOpenByBlock" in app_js and "statum.afterEvaluate(syncPlanRemarks)" in app_js,
           "the queue survives pulse refreshes (maps outside statum re-renders)")
 
     # The rejection payload exactly as the client composes it: the
@@ -671,7 +684,9 @@ def run(ctx):
 
     # (3) The goal page template ships no verdict constants and its
     # script resolves the review block from ANY [data-plan-review]
-    # container via the open-blocks list (the completion-review fix).
+    # container via the open-blocks list (the completion-review fix);
+    # the reject flow itself lives in the shared app.js engine
+    # (app_js fetched in B2 above).
     # Fetched while the review is open so the accept form renders.
     s2_page = http_get_page(auth, "/goals/%d" % s2_goal)
     check("Plan approved. Proceed with implementation." not in s2_page
@@ -681,9 +696,9 @@ def run(ctx):
           "the goal page's accept forms post the verdict marker")
     check("resolveReviewBlock" in s2_page and "goal.open_blocks" in s2_page,
           "the reject flow resolves its block from the open-blocks list (container-keyed)")
-    check("statum.act(block.answer_action, { feedback: feedback })" in s2_page,
+    check("statum.act(block.answer_action, { feedback: feedback })" in app_js,
           "the reject flow posts raw feedback (the server wraps the prefix)")
-    check("extraText.length === 0" in s2_page,
+    check("extraText.length === 0" in app_js,
           "a remark-less rejection with textarea content is allowed (S2)")
 
     # (2) The goal page's reject rides `feedback` (the client composes

test/bundles/pages.py

@@ -894,17 +894,24 @@ def run(ctx):
     # every object M0002 owns exists, so a future in-place edit of
     # M0001 (the regression this goal shipped against) is caught by
     # the serial staying 1 here while the schema drifts.
-    print("== migrations: incremental schema (M0002 recorded) ==")
+    # P1 (M0003) extends the chain: serial 3 records the event ledger
+    # reader indexes.
+    print("== migrations: incremental schema (M0002, M0003 recorded) ==")
     conn = db()
     serials = conn.execute(
         "select serial from __migrations where namespace = 'clanker-saddle'"
         " order by serial").fetchall()
+    event_indexes = {r[0] for r in conn.execute(
+        "select name from sqlite_master"
+        " where type = 'index' and tbl_name = 'events'").fetchall()}
     conn.close()
-    check([s[0] for s in serials] == [1, 2],
-          "the clanker-saddle namespace carries exactly serials 1 and 2"
-          " (a fresh boot ran M0001 then the M0002 no-op)")
-    check(len(serials) == 2,
-          "no duplicate migration rows (M0002's guarded up() applied once)")
+    check([s[0] for s in serials] == [1, 2, 3],
+          "the clanker-saddle namespace carries exactly serials 1, 2 and 3"
+          " (a fresh boot ran M0001 then the M0002 no-op then M0003)")
+    check(len(serials) == 3,
+          "no duplicate migration rows (each guarded up() applied once)")
+    check({"idx_events_scope", "idx_events_terminal"} <= event_indexes,
+          "M0003's event reader indexes exist (fresh boot)")
 
     conn = db()
     tables = {r[0] for r in conn.execute(
@@ -916,3 +923,51 @@ def run(ctx):
           "M0002's tables (notes, stacks, stack_projects) exist")
     check({"prompt_note", "workspace_root"} <= saddle_cols,
           "M0002's saddles columns (prompt_note, workspace_root) exist")
+
+    print("== S5: the task page's LLM usage stat line ==")
+    # usage_summary aggregates the seeded usage events (the daemon's
+    # prompt/completion/elapsed payload) into one muted stat line on
+    # the status card; zero-usage tasks carry no line at all.
+    conn = db()
+    conn.execute("insert into tasks (project_id, title, status, priority,"
+                 " created_at) values (1, 'Usage ledger task', 'working', 5,"
+                 " strftime('%s','now'))")
+    usage_task = conn.execute("select max(id) from tasks").fetchone()[0]
+    for payload in ({"prompt_tokens": 2084, "completion_tokens": 66,
+                     "elapsed_ms": 8519},
+                    {"prompt_tokens": 2458, "completion_tokens": 148,
+                     "elapsed_ms": 6032},
+                    {"prompt_tokens": 4519, "completion_tokens": 243,
+                     "elapsed_ms": 10544}):
+        conn.execute(
+            "insert into events (kind, actor, scope_type, scope_id, payload,"
+            " created_at) values ('usage', 'harness', 'task', ?, ?,"
+            " strftime('%s','now'))",
+            (usage_task, json.dumps(payload)))
+    conn.execute("insert into tasks (project_id, title, status, priority,"
+                 " created_at) values (1, 'Usage magnitude task', 'working', 5,"
+                 " strftime('%s','now'))")
+    magnitude_task = conn.execute("select max(id) from tasks").fetchone()[0]
+    conn.execute(
+        "insert into events (kind, actor, scope_type, scope_id, payload,"
+        " created_at) values ('usage', 'harness', 'task', ?, ?,"
+        " strftime('%s','now'))",
+        (magnitude_task, json.dumps({"prompt_tokens": 45200000,
+                                     "completion_tokens": 1100000,
+                                     "elapsed_ms": 11520000})))
+    conn.commit()
+    conn.close()
+    detail = task_state(auth, usage_task)
+    check(detail.get("usage_line") == "LLM: 3 calls · 9,061 in / 457 out tok · 25s",
+          "the task state aggregates the seeded usage events (calls, tokens,"
+          " wall time): %r" % detail.get("usage_line"))
+    check(task_state(auth, magnitude_task).get("usage_line")
+          == "LLM: 1 calls · 45.2M in / 1.1M out tok · 3h 12m",
+          "token magnitudes and durations format human-friendly"
+          " (%r)" % task_state(auth, magnitude_task).get("usage_line"))
+    task_page = http_get_page(auth, "/tasks/%d" % usage_task)
+    check('stm-if="task.usage_line.length &gt; 0"' in task_page
+          and 'stm-text="task.usage_line"' in task_page,
+          "the task page renders the usage stat line (gated on non-empty)")
+    check(task_state(auth, draft_display).get("usage_line") == "",
+          "a task with no usage events carries no stat line")

test/bundles/replay.py

@@ -296,5 +296,120 @@ def run(ctx):
     check("event.kind == 'comment' ? ' footnote'" in page,
           "the narrative's comment events render with footnote styling")
 
+    print("== P1/P2: event ledger reader indexes + batched ingestion ==")
+    # P1 (i): a FRESH database (this shard's app migrated it at
+    # startup) carries both M0003 reader indexes, and the
+    # scope-filtered reader shapes actually use them — no full
+    # 'SCAN events' for the page/cursor walks or the resume
+    # coalescer's terminal-transition scan.
+    def event_indexes():
+        conn = db()
+        try:
+            return {r[0] for r in conn.execute(
+                "select name from sqlite_master"
+                " where type = 'index' and tbl_name = 'events'").fetchall()}
+        finally:
+            conn.close()
+
+    def query_plan(sql, params=()):
+        conn = db()
+        try:
+            return " ".join(r[3] for r in conn.execute(
+                "explain query plan " + sql, params).fetchall())
+        finally:
+            conn.close()
+
+    check({"idx_events_scope", "idx_events_terminal"} <= event_indexes(),
+          "M0003 creates both event reader indexes on a fresh database")
+    scope_plan = query_plan(
+        "select id from events where scope_type = 'task' and scope_id = ?"
+        " and id < ? order by id desc limit 30", (first, 2 ** 40))
+    check("idx_events_scope" in scope_plan and "SCAN events" not in scope_plan,
+          "the scope-filtered readers search idx_events_scope (no table scan)")
+    terminal_plan = query_plan(
+        "select id, created_at from events where kind = 'state_change'"
+        " and scope_type = 'task' and scope_id = ?", (first,))
+    check("idx_events_terminal" in terminal_plan
+          and "SCAN events" not in terminal_plan,
+          "the terminal-transition reader searches the partial idx_events_terminal")
+
+    # P2 (iii): batched ingestion stays idempotent — a redelivered
+    # batch stores 0 new rows and errors nothing, and a MIXED batch
+    # (one duplicate + one new event) stores exactly the new row.
+    batch_target = open_session_on_fake(first)
+    one_event = [{"seq": 6001, "kind": "progress",
+                  "payload": {"text": "batched"}}]
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": batch_target, "events": one_event})
+    check(status == 200 and int(body.get("stored") or 0) == 1,
+          "a fresh batched event stores one row")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": batch_target,
+                               "events": one_event + [
+                                   {"seq": 6002, "kind": "progress",
+                                    "payload": {"text": "second"}}]})
+    check(status == 200 and int(body.get("stored") or 0) == 1,
+          "a mixed redelivery batch stores only the new row, no error")
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": batch_target, "events": one_event})
+    check(status == 200 and int(body.get("stored") or 0) == 0,
+          "a redelivered batch stores 0 new rows and errors nothing")
+
+    # P2 (iv): a batch containing one invalid event is refused WHOLE
+    # — staging validates before any insert, so nothing partially
+    # stores (the invalid event sits between two valid ones).
+    conn = db()
+    target_rows = conn.execute(
+        "select count(*) from events where session_id = ?",
+        (batch_target,)).fetchone()[0]
+    conn.close()
+    status, body = saddle_api("POST", "/api/saddle/chunks", fake_token,
+                              {"session_id": batch_target, "events": [
+                                  {"seq": 6101, "kind": "progress",
+                                   "payload": {}},
+                                  {"seq": 0, "kind": "progress",
+                                   "payload": {}},
+                                  {"seq": 6102, "kind": "progress",
+                                   "payload": {}}]})
+    check(status == 400, "a batch with an invalid event is refused (HTTP 400)")
+    conn = db()
+    after_rows = conn.execute(
+        "select count(*) from events where session_id = ?",
+        (batch_target,)).fetchone()[0]
+    conn.close()
+    check(after_rows == target_rows,
+          "the refused batch stores nothing (no partial storage)")
+
+    # P1 (ii): a serial-2 database upgraded in place gains the
+    # indexes — strip M0003's record and both indexes off this
+    # shard's database, restart the app, and the startup migration
+    # must reapply them over the existing ledger.
+    conn = db()
+    conn.execute("drop index if exists idx_events_scope")
+    conn.execute("drop index if exists idx_events_terminal")
+    conn.execute("delete from __migrations"
+                 " where namespace = 'clanker-saddle' and serial = 3")
+    conn.commit()
+    conn.close()
+    app.terminate()
+    app.wait(timeout=10)
+    app = start_app()
+    auth = login("admin", "adminpass123")
+    ctx.app = app
+    ctx.auth = auth
+    conn = db()
+    serial3 = conn.execute(
+        "select count(*) from __migrations"
+        " where namespace = 'clanker-saddle' and serial = 3").fetchone()[0]
+    kept_rows = conn.execute(
+        "select count(*) from events where session_id = ?",
+        (batch_target,)).fetchone()[0]
+    conn.close()
+    check({"idx_events_scope", "idx_events_terminal"} <= event_indexes()
+          and serial3 == 1,
+          "a serial-2 database upgraded in place gains both indexes")
+    check(kept_rows == after_rows,
+          "the in-place upgrade leaves the ledger's rows untouched")
+
     # Drain: nothing queued remains for the shard's shutdown.
     check(queued_count() == 0, "the replay shard drains its queue")