perf: reduce CPU usage in event replay, projections, and sidebar

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -102,6 +102,7 @@ describe("OrchestrationEngine", () => {
           return savedEvent;
         }),
       readFromSequence: () => Stream.empty,
+      readAggregateFromSequence: () => Stream.empty,
       readAll: () =>
         Stream.fail(
           new PersistenceSqlError({
@@ -772,6 +773,16 @@ describe("OrchestrationEngine", () => {
       readFromSequence(sequenceExclusive) {
         return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
       },
+      readAggregateFromSequence(aggregateKind, aggregateId, sequenceExclusive) {
+        return Stream.fromIterable(
+          events.filter(
+            (event) =>
+              event.aggregateKind === aggregateKind &&
+              event.aggregateId === aggregateId &&
+              event.sequence > sequenceExclusive,
+          ),
+        );
+      },
       readAll() {
         return Stream.fromIterable(events);
       },
@@ -1004,6 +1015,16 @@ describe("OrchestrationEngine", () => {
       readFromSequence(sequenceExclusive) {
         return Stream.fromIterable(events.filter((event) => event.sequence > sequenceExclusive));
       },
+      readAggregateFromSequence(aggregateKind, aggregateId, sequenceExclusive) {
+        return Stream.fromIterable(
+          events.filter(
+            (event) =>
+              event.aggregateKind === aggregateKind &&
+              event.aggregateId === aggregateId &&
+              event.sequence > sequenceExclusive,
+          ),
+        );
+      },
       readAll() {
         return Stream.fromIterable(events);
       },

apps/server/src/orchestration/Layers/OrchestrationEngine.ts

@@ -309,6 +309,14 @@ const makeOrchestrationEngine = Effect.gen(function* () {
   const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
     eventStore.readFromSequence(fromSequenceExclusive, limit);
 
+  const readAggregateEvents: OrchestrationEngineShape["readAggregateEvents"] = (
+    aggregateKind,
+    aggregateId,
+    fromSequenceExclusive,
+    limit,
+  ) =>
+    eventStore.readAggregateFromSequence(aggregateKind, aggregateId, fromSequenceExclusive, limit);
+
   const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
     Effect.gen(function* () {
       const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
@@ -322,6 +330,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
 
   return {
     readEvents,
+    readAggregateEvents,
     dispatch,
     // Each access creates a fresh PubSub subscription so that multiple
     // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.)

apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts

@@ -2013,6 +2013,107 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
     }),
   );
 
+  it.effect("updates shell summary fields incrementally for message events", () =>
+    Effect.gen(function* () {
+      const projectionPipeline = yield* OrchestrationProjectionPipeline;
+      const eventStore = yield* OrchestrationEventStore;
+      const sql = yield* SqlClient.SqlClient;
+      const appendAndProject = (event: Parameters<typeof eventStore.append>[0]) =>
+        eventStore
+          .append(event)
+          .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent)));
+
+      yield* appendAndProject({
+        type: "thread.created",
+        eventId: EventId.make("evt-incremental-1"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-incremental"),
+        occurredAt: "2026-02-26T14:00:00.000Z",
+        commandId: CommandId.make("cmd-incremental-1"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-incremental-1"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-incremental"),
+          projectId: ProjectId.make("project-incremental"),
+          title: "Thread Incremental",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          runtimeMode: "full-access",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          createdAt: "2026-02-26T14:00:00.000Z",
+          updatedAt: "2026-02-26T14:00:00.000Z",
+        },
+      });
+
+      yield* appendAndProject({
+        type: "thread.message-sent",
+        eventId: EventId.make("evt-incremental-2"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-incremental"),
+        occurredAt: "2026-02-26T14:00:01.000Z",
+        commandId: CommandId.make("cmd-incremental-2"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-incremental-2"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-incremental"),
+          messageId: MessageId.make("message-incremental-user"),
+          role: "user",
+          text: "Do the thing",
+          turnId: null,
+          streaming: false,
+          createdAt: "2026-02-26T14:00:01.000Z",
+          updatedAt: "2026-02-26T14:00:01.000Z",
+        },
+      });
+
+      // A streaming assistant chunk must not move updatedAt (no shell fan-out).
+      yield* appendAndProject({
+        type: "thread.message-sent",
+        eventId: EventId.make("evt-incremental-3"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-incremental"),
+        occurredAt: "2026-02-26T14:00:02.000Z",
+        commandId: CommandId.make("cmd-incremental-3"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-incremental-3"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-incremental"),
+          messageId: MessageId.make("message-incremental-assistant"),
+          role: "assistant",
+          text: "Work",
+          turnId: null,
+          streaming: true,
+          createdAt: "2026-02-26T14:00:02.000Z",
+          updatedAt: "2026-02-26T14:00:02.000Z",
+        },
+      });
+
+      const threadRows = yield* sql<{
+        readonly latestUserMessageAt: string | null;
+        readonly updatedAt: string;
+      }>`
+        SELECT
+          latest_user_message_at AS "latestUserMessageAt",
+          updated_at AS "updatedAt"
+        FROM projection_threads
+        WHERE thread_id = 'thread-incremental'
+      `;
+      assert.deepEqual(threadRows, [
+        {
+          latestUserMessageAt: "2026-02-26T14:00:01.000Z",
+          updatedAt: "2026-02-26T14:00:01.000Z",
+        },
+      ]);
+    }),
+  );
+
   it.effect("ignores non-stale provider approval response failures", () =>
     Effect.gen(function* () {
       const projectionPipeline = yield* OrchestrationProjectionPipeline;

apps/server/src/orchestration/Layers/ProjectionPipeline.ts

@@ -176,6 +176,20 @@ function derivePendingUserInputCountFromActivities(
   return openRequestIds.size;
 }
 
+// Activity kinds that can change the shell summary's pending-approval or
+// pending-user-input counters. All other activity kinds (command output,
+// file edits, streaming progress, ...) leave the summary untouched, so they
+// must not trigger the history-wide summary rebuild — and the shell stream
+// (ws.ts toShellStreamEvent) skips re-broadcasting the thread row for them.
+export const SHELL_SUMMARY_ACTIVITY_KINDS: ReadonlySet<string> = new Set([
+  "approval.requested",
+  "approval.resolved",
+  "provider.approval.respond.failed",
+  "user-input.requested",
+  "user-input.resolved",
+  "provider.user-input.respond.failed",
+]);
+
 function deriveHasActionableProposedPlan(input: {
   readonly latestTurnId: string | null;
   readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;
@@ -713,9 +727,61 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
           return;
         }
 
-        case "thread.message-sent":
+        case "thread.message-sent": {
+          const existingRow = yield* projectionThreadRepository.getById({
+            threadId: event.payload.threadId,
+          });
+          if (Option.isNone(existingRow)) {
+            return;
+          }
+          // A message can only move latestUserMessageAt forward; the other
+          // summary fields depend on activities/plans/approvals, which this
+          // event cannot touch. Update incrementally instead of re-reading the
+          // whole message/plan/activity/approval history per (streamed) chunk.
+          const latestUserMessageAt =
+            event.payload.role === "user" &&
+            (existingRow.value.latestUserMessageAt === null ||
+              event.payload.createdAt > existingRow.value.latestUserMessageAt)
+              ? event.payload.createdAt
+              : existingRow.value.latestUserMessageAt;
+          yield* projectionThreadRepository.upsert({
+            ...existingRow.value,
+            latestUserMessageAt,
+            // Streaming assistant chunks arrive many times per second and
+            // change no shell-visible field; leaving updatedAt untouched lets
+            // the shell stream skip per-chunk upsert fan-out (toShellStreamEvent).
+            updatedAt:
+              event.payload.streaming && event.payload.role === "assistant"
+                ? existingRow.value.updatedAt
+                : event.occurredAt,
+          });
+          return;
+        }
+
+        case "thread.activity-appended": {
+          // Non-summary activities (command output, file edits, progress, ...)
+          // change no shell-visible field: the sidebar timestamp prefers
+          // latestUserMessageAt and status pills derive from session/approval
+          // state. Skip the row churn entirely so high-volume activity streams
+          // don't bump updatedAt (and fan out shell upserts) per event.
+          if (!SHELL_SUMMARY_ACTIVITY_KINDS.has(event.payload.activity.kind)) {
+            return;
+          }
+          const existingRow = yield* projectionThreadRepository.getById({
+            threadId: event.payload.threadId,
+          });
+          if (Option.isNone(existingRow)) {
+            return;
+          }
+          yield* projectionThreadRepository.upsert({
+            ...existingRow.value,
+            updatedAt: event.occurredAt,
+          });
+          yield* refreshThreadShellSummary(event.payload.threadId);
+          return;
+        }
+
         case "thread.proposed-plan-upserted":
-        case "thread.activity-appended":
         case "thread.approval-response-requested":
         case "thread.user-input-response-requested": {
           const existingRow = yield* projectionThreadRepository.getById({

apps/server/src/orchestration/Services/OrchestrationEngine.ts

@@ -37,6 +37,18 @@ export interface OrchestrationEngineShape {
     limit?: number,
   ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
 
+  /**
+   * Replay a single aggregate's persisted events from an exclusive sequence
+   * cursor. Reads via the aggregate stream index, so per-thread catch-up does
+   * not scan the global event range after the cursor.
+   */
+  readonly readAggregateEvents: (
+    aggregateKind: OrchestrationEvent["aggregateKind"],
+    aggregateId: string,
+    fromSequenceExclusive: number,
+    limit?: number,
+  ) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
+
   /**
    * Dispatch a validated orchestration command.
    *

apps/server/src/persistence/Layers/OrchestrationEventStore.test.ts

@@ -69,6 +69,51 @@ layer("OrchestrationEventStore", (it) => {
     }),
   );
 
+  it.effect("replays only the requested aggregate's events after the cursor", () =>
+    Effect.gen(function* () {
+      const eventStore = yield* OrchestrationEventStore;
+      const now = "2026-01-01T00:00:00.000Z";
+
+      const appendProjectCreated = (suffix: string) =>
+        eventStore.append({
+          type: "project.created",
+          eventId: EventId.make(`evt-aggregate-${suffix}`),
+          aggregateKind: "project",
+          aggregateId: ProjectId.make(`project-${suffix}`),
+          occurredAt: now,
+          commandId: CommandId.make(`cmd-aggregate-${suffix}`),
+          causationEventId: null,
+          correlationId: null,
+          metadata: {},
+          payload: {
+            projectId: ProjectId.make(`project-${suffix}`),
+            title: `Project ${suffix}`,
+            workspaceRoot: `/tmp/project-${suffix}`,
+            defaultModelSelection: null,
+            scripts: [],
+            createdAt: now,
+            updatedAt: now,
+          },
+        });
+
+      const first = yield* appendProjectCreated("a");
+      yield* appendProjectCreated("b");
+
+      const replayed = yield* Stream.runCollect(
+        eventStore.readAggregateFromSequence("project", first.aggregateId, 0),
+      ).pipe(Effect.map((chunk) => Array.from(chunk)));
+      assert.deepStrictEqual(
+        replayed.map((event) => event.aggregateId),
+        [first.aggregateId],
+      );
+
+      const afterCursor = yield* Stream.runCollect(
+        eventStore.readAggregateFromSequence("project", first.aggregateId, first.sequence),
+      ).pipe(Effect.map((chunk) => Array.from(chunk)));
+      assert.deepStrictEqual(afterCursor, []);
+    }),
+  );
+
   it.effect("fails with PersistenceDecodeError when stored json is invalid", () =>
     Effect.gen(function* () {
       const eventStore = yield* OrchestrationEventStore;