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

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

@@ -3002,8 +3002,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 171).
  */
 const EXCLUDED_FROM_PAGINATION_SQL_LIST = toSqlStringList([
   ...HIDDEN_SYSTEM_SUBTYPES,
@@ -3016,14 +3019,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 = ?
@@ -3038,9 +3039,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

@@ -1211,7 +1211,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;