perf(daemon): make sdk_messages subtype filters sargable (#2330)

packages/daemon/src/lib/rpc-handlers/live-query-handlers.ts

@@ -766,7 +766,7 @@ sdk_rows AS (
   WHERE (sm.message_type != 'user' OR COALESCE(sm.send_status, 'consumed') IN ('consumed', 'failed'))
     AND (
       sm.message_type != 'system'
-      OR COALESCE(sm.message_subtype, '') != 'informational'
+      OR sm.message_subtype_norm != 'informational'
       OR NOT json_valid(sm.sdk_message)
       OR COALESCE(
         CASE
@@ -777,7 +777,7 @@ sdk_rows AS (
     )
     AND (
       sm.message_type != 'system'
-      OR COALESCE(sm.message_subtype, '') != 'worker_shutting_down'
+      OR sm.message_subtype_norm != 'worker_shutting_down'
       OR NOT EXISTS (
         SELECT 1
         FROM sdk_messages newer
@@ -1831,7 +1831,7 @@ sdk_rows_raw AS (
   WHERE (sm.message_type != 'user' OR COALESCE(sm.send_status, 'consumed') IN ('consumed', 'failed'))
     AND (
       sm.message_type != 'system'
-      OR COALESCE(sm.message_subtype, '') != 'informational'
+      OR sm.message_subtype_norm != 'informational'
       OR NOT json_valid(sm.sdk_message)
       OR COALESCE(
         CASE
@@ -1842,7 +1842,7 @@ sdk_rows_raw AS (
     )
     AND (
       sm.message_type != 'system'
-      OR COALESCE(sm.message_subtype, '') != 'worker_shutting_down'
+      OR sm.message_subtype_norm != 'worker_shutting_down'
       OR NOT EXISTS (
         SELECT 1
         FROM sdk_messages newer
@@ -3017,7 +3017,7 @@ SELECT
   (SELECT COUNT(*) FROM sdk_messages sm WHERE sm.session_id = s.id
     AND sm.parent_tool_use_id IS NULL
     AND (sm.message_type != 'user' OR COALESCE(sm.send_status, 'consumed') IN ('consumed', 'failed'))
-    AND COALESCE(sm.message_subtype, '') NOT IN (${EXCLUDED_FROM_PAGINATION_SQL_LIST})) as messageCount,
+    AND sm.message_subtype_norm NOT IN (${EXCLUDED_FROM_PAGINATION_SQL_LIST})) as messageCount,
   (unixepoch(s.last_active_at) - 0) * 1000 as lastActiveAt
 FROM sessions s
 INNER JOIN spaces sp ON sp.id = ?
@@ -3078,7 +3078,7 @@ function toSqlStringList(subtypes: Iterable<string>): string {
 
 const BACKGROUND_TASK_METADATA_SQL_LIST = toSqlStringList(BACKGROUND_TASK_METADATA_SUBTYPES);
 
-const BACKGROUND_TASK_METADATA_SQL = `
+export const BACKGROUND_TASK_METADATA_SQL = `
 WITH recent_metadata AS (
   SELECT
     id,
@@ -3094,7 +3094,7 @@ WITH recent_metadata AS (
   FROM sdk_messages
   WHERE session_id = ?
     AND parent_tool_use_id IS NULL
-    AND COALESCE(message_subtype, '') IN (${BACKGROUND_TASK_METADATA_SQL_LIST})
+    AND message_subtype_norm IN (${BACKGROUND_TASK_METADATA_SQL_LIST})
   ORDER BY timestamp DESC, rowid DESC
   LIMIT ${BACKGROUND_TASK_METADATA_BATCH_SIZE}
 ),
@@ -3118,7 +3118,7 @@ task_starts AS (
   FROM sdk_messages
   WHERE session_id = ?
     AND parent_tool_use_id IS NULL
-    AND COALESCE(message_subtype, '') = 'task_started'
+    AND message_subtype_norm = 'task_started'
     AND COALESCE(
       CASE WHEN json_valid(sdk_message) THEN json_extract(sdk_message, '$.task_id') END,
       task_id
@@ -3156,7 +3156,7 @@ latest_progress AS (
     FROM sdk_messages
     WHERE session_id = ?
       AND parent_tool_use_id IS NULL
-      AND COALESCE(message_subtype, '') = 'task_progress'
+      AND message_subtype_norm = 'task_progress'
       AND COALESCE(
         CASE WHEN json_valid(sdk_message) THEN json_extract(sdk_message, '$.tool_use_id') END,
         ''
@@ -3193,10 +3193,10 @@ WITH top_level AS (
   WHERE session_id = ?1
     AND parent_tool_use_id IS NULL
     AND (message_type != 'user' OR COALESCE(send_status, 'consumed') IN ('consumed', 'failed'))
-    AND COALESCE(message_subtype, '') NOT IN (${EXCLUDED_FROM_PAGINATION_SQL_LIST})
+    AND message_subtype_norm NOT IN (${EXCLUDED_FROM_PAGINATION_SQL_LIST})
     AND (
       message_type != 'system'
-      OR COALESCE(message_subtype, '') != 'informational'
+      OR message_subtype_norm != 'informational'
       OR NOT json_valid(sdk_message)
       OR COALESCE(
         CASE
@@ -3207,7 +3207,7 @@ WITH top_level AS (
     )
     AND (
       message_type != 'system'
-      OR COALESCE(message_subtype, '') != 'worker_shutting_down'
+      OR message_subtype_norm != 'worker_shutting_down'
       OR NOT EXISTS (
         SELECT 1
         FROM sdk_messages newer
@@ -3256,7 +3256,7 @@ subagent AS (
   FROM sdk_messages sm
   WHERE sm.session_id = ?1
     AND sm.parent_tool_use_id IN (SELECT id FROM tool_use_ids)
-    AND COALESCE(sm.message_subtype,'') != 'thinking_tokens'
+    AND sm.message_subtype_norm != 'thinking_tokens'
     AND (sm.message_type != 'user' OR COALESCE(sm.send_status, 'consumed') IN ('consumed', 'failed'))
 )
 SELECT

packages/daemon/src/storage/schema/index.ts

@@ -195,6 +195,11 @@ export function createTables(db: BunDatabase): void {
         session_id TEXT NOT NULL,
         message_type TEXT NOT NULL,
         message_subtype TEXT,
+        -- VIRTUAL generated column so subtype equality/IN filters (which must
+        -- treat NULL as '') become sargable against idx_sdk_messages_session_subtype_parent.
+        -- VIRTUAL = computed on read, no storage/rewrite; semantically identical
+        -- to COALESCE(message_subtype,'') by construction.
+        message_subtype_norm TEXT GENERATED ALWAYS AS (COALESCE(message_subtype, '')) VIRTUAL,
         sdk_message TEXT NOT NULL,
         timestamp TEXT NOT NULL,
         send_status TEXT DEFAULT 'consumed' CHECK(send_status IN ('deferred', 'enqueued', 'consumed', 'failed')),
@@ -932,6 +937,13 @@ function createIndexes(db: BunDatabase): void {
       ON sdk_messages(session_id, send_status, json_extract(sdk_message, '$.uuid'))`);
   db.exec(`CREATE INDEX IF NOT EXISTS idx_sdk_messages_type
       ON sdk_messages(message_type, message_subtype)`);
+  // Makes the chat-view subtype filters sargable: the messages.bySession
+  // background-task sidecar and friends constrain (session_id, subtype, parent_tool_use_id),
+  // and ~78% of rows have message_subtype NULL, so filtering on the raw column
+  // forces a non-sargable COALESCE. message_subtype_norm is the NULL-coalesced
+  // generated column; this composite turns those filters into index seeks.
+  db.exec(`CREATE INDEX IF NOT EXISTS idx_sdk_messages_session_subtype_parent
+      ON sdk_messages(session_id, message_subtype_norm, parent_tool_use_id)`);
   db.exec(`CREATE INDEX IF NOT EXISTS idx_sdk_messages_send_status
       ON sdk_messages(session_id, send_status)`);
   // Task-scoped feeds and activity views read directly from this column.

packages/daemon/src/storage/schema/migrations.ts

@@ -793,6 +793,17 @@ export function runMigrations(db: BunDatabase, createBackup: () => void): void {
   // lookup. (The live-query task-scope nodeExecStmt already drives off
   // idx_node_executions_run and is unaffected.)
   run(migrationMarkerKey(168), () => runMigration168(db));
+
+  // Migration 169: Add a VIRTUAL generated column message_subtype_norm =
+  // COALESCE(message_subtype,'') plus a (session_id, message_subtype_norm,
+  // parent_tool_use_id) index, so the chat-view subtype filters (the
+  // messages.bySession background-task sidecar and friends) become sargable
+  // instead of seeking to all top-level rows and filtering row-by-row. VIRTUAL
+  // makes the ALTER schema-only (no table rewrite) — critical for the 15GB
+  // production DB. New databases get both via createTables(); this brings
+  // existing databases up to parity. (Originally M168 on this branch;
+  // renumbered to M169 because dev shipped M168 for node_executions(agent_session_id) in #2343.)
+  run(migrationMarkerKey(169), () => runMigration169(db));
 }
 
 function migrationMarkerKey(version: number): string {
@@ -11477,3 +11488,48 @@ export function runMigration168(db: BunDatabase): void {
     ON node_executions(agent_session_id)
   `);
 }
+
+/**
+ * Migration 169: Add the `message_subtype_norm` VIRTUAL generated column and a
+ * `(session_id, message_subtype_norm, parent_tool_use_id)` index.
+ *
+ * ~78% of `sdk_messages` rows have `message_subtype IS NULL`, so the chat-view
+ * subtype filters wrap the column as `COALESCE(message_subtype,'') = …`, which
+ * is non-sargable and forces the `(session_id, parent_tool_use_id)` index to
+ * seek to every top-level row before filtering — a 2.35s stall on the 167k-row
+ * coder session (see issue #2330). The generated column makes those filters
+ * sargable against the new composite index.
+ *
+ * `VIRTUAL` (not `STORED`) keeps the `ALTER TABLE` schema-only — no table
+ * rewrite, so it is safe on the 15GB production DB. SQLite 3.51 (bun:sqlite's
+ * bundled version) supports indexing VIRTUAL generated columns. New databases
+ * already get both the column and index from `createTables()`; this migration
+ * brings existing databases to parity. Idempotent.
+ *
+ * (Originally M168 on this branch; renumbered to M169 because dev shipped M168
+ * for node_executions(agent_session_id) in #2343.)
+ */
+export function runMigration169(db: BunDatabase): void {
+  if (!tableExists(db, 'sdk_messages')) return;
+
+  // Generated columns are hidden from `pragma_table_info` (the shared
+  // tableHasColumn helper), so check via `pragma_table_xinfo` instead —
+  // otherwise the guard misses the column and a second run errors with
+  // "duplicate column name".
+  const hasNormColumn = !!db
+    .prepare(
+      `SELECT name FROM pragma_table_xinfo('sdk_messages') WHERE name = 'message_subtype_norm'`
+    )
+    .get();
+  if (!hasNormColumn) {
+    db.exec(`
+      ALTER TABLE sdk_messages
+        ADD COLUMN message_subtype_norm TEXT GENERATED ALWAYS AS (COALESCE(message_subtype, '')) VIRTUAL
+    `);
+  }
+
+  db.exec(`
+    CREATE INDEX IF NOT EXISTS idx_sdk_messages_session_subtype_parent
+    ON sdk_messages(session_id, message_subtype_norm, parent_tool_use_id)
+  `);
+}

packages/daemon/tests/unit/2-handlers/rpc-handlers/live-query-handlers.test.ts

@@ -4411,6 +4411,7 @@ describe('NAMED_QUERY_REGISTRY', () => {
             session_id TEXT,
             message_type TEXT,
             message_subtype TEXT,
+            message_subtype_norm TEXT GENERATED ALWAYS AS (COALESCE(message_subtype, '')) VIRTUAL,
             send_status TEXT,
             parent_tool_use_id TEXT,
             timestamp TEXT

packages/daemon/tests/unit/2-handlers/rpc-handlers/live-query-messages.test.ts

@@ -18,6 +18,7 @@ import { createTables } from '../../../../src/storage/schema';
 import { createReactiveDatabase } from '../../../../src/storage/reactive-database';
 import { LiveQueryEngine } from '../../../../src/storage/live-query';
 import {
+  BACKGROUND_TASK_METADATA_SQL,
   NAMED_QUERY_REGISTRY,
   setupLiveQueryHandlers,
 } from '../../../../src/lib/rpc-handlers/live-query-handlers';
@@ -795,6 +796,68 @@ describe('messages.bySession — SQL behavior', () => {
     expect(plan).toContain('idx_sdk_messages_parent_tool_use_id');
   });
 
+  test('background-task sidecar subtype filter is sargable via message_subtype_norm', () => {
+    // See issue #2330: 78% of rows have message_subtype NULL, so a
+    // COALESCE(message_subtype,'') filter is non-sargable and forced the
+    // (session_id, parent_tool_use_id) index to seek every top-level row. The
+    // message_subtype_norm generated column + composite index must turn the
+    // sidecar's subtype IN filter into an index seek instead.
+    for (const subtype of ['task_started', 'task_updated', 'task_notification']) {
+      insertSdkMessage(db, {
+        id: `bg-${subtype}`,
+        sessionId: 's1',
+        messageType: 'system',
+        messageSubtype: subtype,
+        sdkMessage: { type: 'system', subtype, uuid: `${subtype}-uuid`, session_id: 's1' },
+        timestamp: '2024-01-01 00:00:01',
+      });
+    }
+    insertSdkMessage(db, {
+      id: 'null-subtype',
+      sessionId: 's1',
+      messageType: 'system',
+      messageSubtype: null,
+      sdkMessage: { type: 'system', uuid: 'null-uuid', session_id: 's1' },
+      timestamp: '2024-01-01 00:00:00',
+    });
+
+    // The sidecar has three session_id=? parameters (one per CTE).
+    const planRows = db
+      .prepare(`EXPLAIN QUERY PLAN ${BACKGROUND_TASK_METADATA_SQL}`)
+      .all('s1', 's1', 's1') as Array<{ detail: string }>;
+    const plan = planRows.map((row) => row.detail).join('\n');
+    // The load-bearing assertion: the new index drives the subtype predicate.
+    expect(plan).toContain('idx_sdk_messages_session_subtype_parent');
+    expect(plan).toContain('message_subtype_norm');
+    // ...and it resolves to a seek (SEARCH), not a scan of sdk_messages.
+    expect(plan).toContain('SEARCH');
+  });
+
+  test('message_subtype_norm equals COALESCE(message_subtype, "") for every row', () => {
+    insertSdkMessage(db, {
+      id: 'null',
+      sessionId: 's1',
+      messageType: 'system',
+      messageSubtype: null,
+      sdkMessage: { type: 'system', uuid: 'n', session_id: 's1' },
+    });
+    insertSdkMessage(db, {
+      id: 'progress',
+      sessionId: 's1',
+      messageType: 'system',
+      messageSubtype: 'task_progress',
+      sdkMessage: { type: 'system', uuid: 'p', session_id: 's1' },
+    });
+
+    const drift = db
+      .prepare(
+        `SELECT COUNT(*) AS n FROM sdk_messages
+          WHERE message_subtype_norm != COALESCE(message_subtype, '')`
+      )
+      .get() as { n: number };
+    expect(drift.n).toBe(0);
+  });
+
   test('includes subagent messages whose parent_tool_use_id matches a top-level tool_use', () => {
     // Top-level assistant row with a tool_use in its content.
     insertSdkMessage(db, {