feat(space): maintain sessions.visible_message_count counter [#2336]

packages/daemon/scripts/recover-messages.ts

@@ -20,6 +20,7 @@
 
 import { readFileSync } from 'fs';
 import { Database } from '../src/storage/sqlite-compat';
+import { SDKMessageRepository } from '../src/storage/repositories/sdk-message-repository';
 
 const dbPath = process.argv[2];
 
@@ -159,8 +160,15 @@ const insertMessage = db.prepare(`
 `);
 
 let messagesInserted = 0;
+// Every matched session is recomputed on every run, even if this invocation
+// inserts no new rows for it: a prior interrupted run may have inserted rows
+// but exited before Step 6, leaving the counter stale, and the existing-UUID
+// `continue` below would then skip the session entirely. recomputeVisibleMessageCount
+// is idempotent, so recomputing all matched sessions each run is safe.
+const touchedSessions = new Set<string>();
 
 for (const [kaiSessionId, msgs] of messagesByKaiSession.entries()) {
+  touchedSessions.add(kaiSessionId);
   for (const msg of msgs) {
     if (existingMessageIds.has(msg.uuid)) continue;
 
@@ -280,6 +288,7 @@ for (const [sdkSessionId, msgs] of orphansBySDKSession.entries()) {
       null
     );
     orphanSessionsCreated++;
+    touchedSessions.add(newSessionId);
 
     // Insert messages for this new session
     for (const msg of msgs) {
@@ -306,6 +315,25 @@ for (const [sdkSessionId, msgs] of orphansBySDKSession.entries()) {
 console.log(`   Orphan sessions created: ${orphanSessionsCreated}`);
 console.log(`   Orphan messages restored: ${orphanMessagesInserted}`);
 
+// Step 6: Recompute visible_message_count for every touched session.
+console.log('\nStep 6: Recomputing visible_message_count for recovered sessions...');
+// The inserts above bypass SDKMessageRepository, so reuse its shared badge
+// predicate to recompute the maintained counter (no-op on a schema that
+// doesn't carry the visible_message_count column yet — the migration backfill
+// covers that). Deliberately no hardcoded migration number here so this comment
+// survives future renumbers.
+const recoverRepo = new SDKMessageRepository(db);
+let recomputed = 0;
+for (const sid of touchedSessions) {
+  try {
+    recoverRepo.recomputeVisibleMessageCount(sid);
+    recomputed++;
+  } catch {
+    // Skip recompute errors
+  }
+}
+console.log(`   Recomputed ${recomputed} session(s)`);
+
 // Final summary
 const finalSessions = db.query('SELECT COUNT(*) as count FROM sessions').get() as { count: number };
 const finalMessages = db.query('SELECT COUNT(*) as count FROM sdk_messages').get() as {

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

@@ -2781,8 +2781,11 @@ function mapSessionRow(row: Record<string, unknown>): Record<string, unknown> {
 
 /**
  * Render-hidden rows excluded before applying transcript pagination limits.
- * Shared by `messages.bySession` and the `spaceSessions.bySpace` messageCount
- * subquery so the unread count only reflects visible top-level rows.
+ * Used by `messages.bySession` to cap the visible transcript window. The
+ * `spaceSessions.bySpace` badge count now reads the maintained
+ * `sessions.visible_message_count` column instead of a correlated COUNT(*), but
+ * the same visibility predicate is enforced there incrementally by
+ * SDKMessageRepository (and backfilled by migration 177).
  */
 const EXCLUDED_FROM_PAGINATION_SQL_LIST = toSqlStringList([
   ...HIDDEN_SYSTEM_SUBTYPES,
@@ -2795,14 +2798,12 @@ SELECT
   s.title as title,
   s.status as status,
   s.processing_state as processingState,
-  -- Mirror the messages.bySession top-level visibility predicate so the count
-  -- reflects what the user actually sees: top-level rows only (subagent rows
-  -- ride on their parent turn), non-deferred user rows, and non-hidden
-  -- subtypes. A best-effort approximation of the visible transcript.
-  (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 sm.message_subtype_norm NOT IN (${EXCLUDED_FROM_PAGINATION_SQL_LIST})) as messageCount,
+  -- Read the maintained counter directly instead of a correlated COUNT(*) over
+  -- sdk_messages per session (previously ~92ms warm for dev-neokai, re-run every
+  -- 150ms debounce). SDKMessageRepository keeps this in sync with the same
+  -- visibility predicate (top-level rows, non-deferred user rows, non-hidden
+  -- subtypes) on every sdk_messages mutation.
+  s.visible_message_count as messageCount,
   (unixepoch(s.last_active_at) - 0) * 1000 as lastActiveAt
 FROM sessions s
 INNER JOIN spaces sp ON sp.id = ?
@@ -2817,9 +2818,9 @@ ORDER BY s.last_active_at DESC, s.id DESC
  * `processingState` is the persisted JSON-serialised `AgentProcessingState`
  * (mirrors `mapSessionRow` for the global sessions list); the web client parses
  * it via `session-status.ts`'s `parseProcessingState`. `messageCount` is the
- * per-session visible SDK message total (deferred/enqueued user rows are
- * excluded, matching the `messages.bySession` transcript view), coerced to a
- * number so the sidebar can drive an unread badge the same way global chat
+ * maintained `sessions.visible_message_count` counter (deferred/enqueued user
+ * rows are excluded, matching the `messages.bySession` transcript view), coerced
+ * to a number so the sidebar can drive an unread badge the same way global chat
  * sessions do.
  */
 function mapSpaceSessionRow(row: Record<string, unknown>): Record<string, unknown> {

packages/daemon/src/lib/space/runtime/space-runtime.ts

@@ -1103,7 +1103,7 @@ export class SpaceRuntime {
    */
   private getSdkMessageRepo(): SDKMessageRepository {
     if (!this.sdkMessageRepo) {
-      this.sdkMessageRepo = new SDKMessageRepository(this.config.db);
+      this.sdkMessageRepo = new SDKMessageRepository(this.config.db, this.config.reactiveDb);
     }
     return this.sdkMessageRepo;
   }

packages/daemon/src/storage/index.ts

@@ -140,7 +140,7 @@ export class Database {
     this.shortIdAllocator = new ShortIdAllocator(db);
     const shortIdAllocator = this.shortIdAllocator;
     this.sessionRepo = new SessionRepository(db);
-    this.sdkMessageRepo = new SDKMessageRepository(db);
+    this.sdkMessageRepo = new SDKMessageRepository(db, reactiveDb);
     this.settingsRepo = new SettingsRepository(db);
     this.githubMappingRepo = new GitHubMappingRepository(db);
     this.inboxItemRepo = new InboxItemRepository(db);

packages/daemon/src/storage/reactive-database.ts

@@ -68,8 +68,11 @@ export interface ReactiveDatabase {
   /**
    * Manually notify that a table has changed.
    * Used for tables whose writes bypass the proxy (e.g., direct SQL via external repos).
+   * Pass a `scope` when known so the change can batch cleanly inside a reactive
+   * transaction — an unscoped notify forces the whole flushed batch to undefined
+   * scope, poisoning properly-scoped writes in the same batch.
    */
-  notifyChange(table: string): void;
+  notifyChange(table: string, scope?: TableChangeScope): void;
 }
 
 // Mapping from facade method name to table name + optional scope extractor.
@@ -408,8 +411,8 @@ export function createReactiveDatabase(db: Database): ReactiveDatabase {
         pendingTableScopes.clear();
       }
     },
-    notifyChange(table: string): void {
-      incrementAndEmit(table);
+    notifyChange(table: string, scope?: TableChangeScope): void {
+      incrementAndEmit(table, scope);
     },
   };
   return reactiveDb as ReactiveDatabase;