fix(stats): keep reverted usage in session stats

packages/core/schema.json

@@ -1,8 +1,8 @@
 {
   "version": "7",
   "dialect": "sqlite",
-  "id": "5067fae0-90ff-4601-82d7-dd050ea9dc8f",
-  "prevIds": ["e98619ef-c334-45e3-8c48-9932ed1ef290"],
+  "id": "9b8d5c43-f59d-4cca-9c90-f66c5bc96060",
+  "prevIds": ["5067fae0-90ff-4601-82d7-dd050ea9dc8f"],
   "ddl": [
     {
       "name": "account_state",
@@ -68,6 +68,10 @@
       "name": "session_pending",
       "entityType": "tables"
     },
+    {
+      "name": "session_revert_usage",
+      "entityType": "tables"
+    },
     {
       "name": "session_v2",
       "entityType": "tables"
@@ -1080,6 +1084,136 @@
       "entityType": "columns",
       "table": "session_pending"
     },
+    {
+      "type": "text",
+      "notNull": false,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "id",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "text",
+      "notNull": true,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "session_id",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "text",
+      "notNull": true,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "type",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "time_created",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "text",
+      "notNull": false,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "model_provider_id",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "text",
+      "notNull": false,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "model_id",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "text",
+      "notNull": false,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "model_variant",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "real",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "cost",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "tokens_input",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "tokens_output",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "tokens_reasoning",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "tokens_cache_read",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
+    {
+      "type": "integer",
+      "notNull": true,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "tokens_cache_write",
+      "entityType": "columns",
+      "table": "session_revert_usage"
+    },
     {
       "type": "text",
       "notNull": false,
@@ -1629,6 +1763,17 @@
       "entityType": "fks",
       "table": "session_pending"
     },
+    {
+      "columns": ["session_id"],
+      "tableTo": "session_v2",
+      "columnsTo": ["id"],
+      "onUpdate": "NO ACTION",
+      "onDelete": "CASCADE",
+      "nameExplicit": false,
+      "name": "fk_session_revert_usage_session_id_session_v2_id_fk",
+      "entityType": "fks",
+      "table": "session_revert_usage"
+    },
     {
       "columns": ["project_id"],
       "tableTo": "project",
@@ -1770,6 +1915,13 @@
       "table": "session_pending",
       "entityType": "pks"
     },
+    {
+      "columns": ["id"],
+      "nameExplicit": false,
+      "name": "session_revert_usage_pk",
+      "table": "session_revert_usage",
+      "entityType": "pks"
+    },
     {
       "columns": ["id"],
       "nameExplicit": false,
@@ -2016,6 +2168,38 @@
       "entityType": "indexes",
       "table": "session_pending"
     },
+    {
+      "columns": [
+        {
+          "value": "session_id",
+          "isExpression": false
+        },
+        {
+          "value": "time_created",
+          "isExpression": false
+        }
+      ],
+      "isUnique": false,
+      "where": null,
+      "origin": "manual",
+      "name": "session_revert_usage_session_time_created_idx",
+      "entityType": "indexes",
+      "table": "session_revert_usage"
+    },
+    {
+      "columns": [
+        {
+          "value": "time_created",
+          "isExpression": false
+        }
+      ],
+      "isUnique": false,
+      "where": null,
+      "origin": "manual",
+      "name": "session_revert_usage_time_created_idx",
+      "entityType": "indexes",
+      "table": "session_revert_usage"
+    },
     {
       "columns": [
         {

packages/core/src/database/migration/20261022034137_session_revert_usage.ts

@@ -0,0 +1,36 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration.js"
+
+const migration: DatabaseMigration.Migration = {
+  id: "20261022034137_session_revert_usage",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`
+        CREATE TABLE \`session_revert_usage\` (
+          \`id\` text PRIMARY KEY,
+          \`session_id\` text NOT NULL,
+          \`type\` text NOT NULL,
+          \`time_created\` integer NOT NULL,
+          \`model_provider_id\` text,
+          \`model_id\` text,
+          \`model_variant\` text,
+          \`cost\` real DEFAULT 0 NOT NULL,
+          \`tokens_input\` integer DEFAULT 0 NOT NULL,
+          \`tokens_output\` integer DEFAULT 0 NOT NULL,
+          \`tokens_reasoning\` integer DEFAULT 0 NOT NULL,
+          \`tokens_cache_read\` integer DEFAULT 0 NOT NULL,
+          \`tokens_cache_write\` integer DEFAULT 0 NOT NULL,
+          CONSTRAINT \`fk_session_revert_usage_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE
+        );
+      `)
+      yield* tx.run(
+        `CREATE INDEX \`session_revert_usage_session_time_created_idx\` ON \`session_revert_usage\` (\`session_id\`,\`time_created\`);`,
+      )
+      yield* tx.run(
+        `CREATE INDEX \`session_revert_usage_time_created_idx\` ON \`session_revert_usage\` (\`time_created\`);`,
+      )
+    })
+  },
+}
+
+export default migration

packages/core/src/database/schema.gen.ts

@@ -180,6 +180,24 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
           CONSTRAINT \`fk_session_pending_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE
         );
       `)
+      yield* tx.run(`
+        CREATE TABLE \`session_revert_usage\` (
+          \`id\` text PRIMARY KEY,
+          \`session_id\` text NOT NULL,
+          \`type\` text NOT NULL,
+          \`time_created\` integer NOT NULL,
+          \`model_provider_id\` text,
+          \`model_id\` text,
+          \`model_variant\` text,
+          \`cost\` real DEFAULT 0 NOT NULL,
+          \`tokens_input\` integer DEFAULT 0 NOT NULL,
+          \`tokens_output\` integer DEFAULT 0 NOT NULL,
+          \`tokens_reasoning\` integer DEFAULT 0 NOT NULL,
+          \`tokens_cache_read\` integer DEFAULT 0 NOT NULL,
+          \`tokens_cache_write\` integer DEFAULT 0 NOT NULL,
+          CONSTRAINT \`fk_session_revert_usage_session_id_session_v2_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session_v2\`(\`id\`) ON DELETE CASCADE
+        );
+      `)
       yield* tx.run(`
         CREATE TABLE \`session_v2\` (
           \`id\` text PRIMARY KEY,
@@ -270,6 +288,12 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
       yield* tx.run(
         `CREATE UNIQUE INDEX \`session_pending_session_admitted_seq_idx\` ON \`session_pending\` (\`session_id\`,\`admitted_seq\`);`,
       )
+      yield* tx.run(
+        `CREATE INDEX \`session_revert_usage_session_time_created_idx\` ON \`session_revert_usage\` (\`session_id\`,\`time_created\`);`,
+      )
+      yield* tx.run(
+        `CREATE INDEX \`session_revert_usage_time_created_idx\` ON \`session_revert_usage\` (\`time_created\`);`,
+      )
       yield* tx.run(`CREATE INDEX \`session_v2_project_idx\` ON \`session_v2\` (\`project_id\`);`)
       yield* tx.run(`CREATE INDEX \`session_v2_workspace_idx\` ON \`session_v2\` (\`workspace_id\`);`)
       yield* tx.run(`CREATE INDEX \`session_v2_parent_idx\` ON \`session_v2\` (\`parent_id\`);`)

packages/core/src/session/projector.ts

@@ -793,6 +793,36 @@ const layer = Layer.effectDiscard(
           .get()
           .pipe(Effect.orDie)
         if (!boundary) return yield* Effect.die(new Error(`Revert boundary message not found: ${event.data.to}`))
+        // Preserve the reverted rows' prompts, steps and usage for SessionStats before
+        // deleting them. The fork filter matches SessionStats: rows a fork copied from its
+        // parent belong to the parent's stats, not this session's.
+        yield* db
+          .run(
+            sql`
+              INSERT OR IGNORE INTO session_revert_usage (id, session_id, type, time_created, model_provider_id, model_id, model_variant, cost, tokens_input, tokens_output, tokens_reasoning, tokens_cache_read, tokens_cache_write)
+              SELECT
+                message.id,
+                message.session_id,
+                message.type,
+                message.time_created,
+                json_extract(message.data, '$.model.providerID'),
+                json_extract(message.data, '$.model.id'),
+                json_extract(message.data, '$.model.variant'),
+                coalesce(json_extract(message.data, '$.cost'), 0),
+                coalesce(json_extract(message.data, '$.tokens.input'), 0),
+                coalesce(json_extract(message.data, '$.tokens.output'), 0),
+                coalesce(json_extract(message.data, '$.tokens.reasoning'), 0),
+                coalesce(json_extract(message.data, '$.tokens.cache.read'), 0),
+                coalesce(json_extract(message.data, '$.tokens.cache.write'), 0)
+              FROM ${SessionMessageTable} AS message
+              JOIN ${SessionTable} AS session ON session.id = message.session_id
+              WHERE message.type IN ('user', 'assistant')
+                AND message.session_id = ${event.data.sessionID}
+                AND message.seq >= ${boundary.seq}
+                AND (session.fork_session_id IS NULL OR message.time_created >= session.time_created)
+            `,
+          )
+          .pipe(Effect.orDie)
         yield* db
           .delete(SessionMessageTable)
           .where(

packages/core/src/session/sql.ts

@@ -98,6 +98,35 @@ export const SessionMessageTable = sqliteTable(
   ],
 )
 
+// Messages removed by a revert commit, copied out before the rows are deleted so SessionStats
+// can keep counting the reverted prompts, steps and usage. Written in the projector, which
+// runs regardless of event persistence (unlike durable events, which only workerd stores).
+export const SessionRevertUsageTable = sqliteTable(
+  "session_revert_usage",
+  {
+    id: text().$type<SessionMessage.ID>().primaryKey(),
+    session_id: text()
+      .$type<SessionSchema.ID>()
+      .notNull()
+      .references(() => SessionTable.id, { onDelete: "cascade" }),
+    type: text().$type<SessionMessage.Type>().notNull(),
+    time_created: integer().notNull(),
+    model_provider_id: text(),
+    model_id: text(),
+    model_variant: text(),
+    cost: real().notNull().default(0),
+    tokens_input: integer().notNull().default(0),
+    tokens_output: integer().notNull().default(0),
+    tokens_reasoning: integer().notNull().default(0),
+    tokens_cache_read: integer().notNull().default(0),
+    tokens_cache_write: integer().notNull().default(0),
+  },
+  (table) => [
+    index("session_revert_usage_session_time_created_idx").on(table.session_id, table.time_created),
+    index("session_revert_usage_time_created_idx").on(table.time_created),
+  ],
+)
+
 export const SessionPendingTable = sqliteTable(
   "session_pending",
   {