perf(server): stop reading a thread's whole activity timeline per event

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

@@ -2086,6 +2086,125 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
     }),
   );
 
+  it.effect("counts an unresolved user-input request on the thread shell", () =>
+    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: "project.created",
+        eventId: EventId.make("evt-open-user-input-1"),
+        aggregateKind: "project",
+        aggregateId: ProjectId.make("project-open-user-input"),
+        occurredAt: "2026-02-26T13:00:00.000Z",
+        commandId: CommandId.make("cmd-open-user-input-1"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-open-user-input-1"),
+        metadata: {},
+        payload: {
+          projectId: ProjectId.make("project-open-user-input"),
+          title: "Project Open User Input",
+          workspaceRoot: "/tmp/project-open-user-input",
+          defaultModelSelection: null,
+          scripts: [],
+          createdAt: "2026-02-26T13:00:00.000Z",
+          updatedAt: "2026-02-26T13:00:00.000Z",
+        },
+      });
+
+      yield* appendAndProject({
+        type: "thread.created",
+        eventId: EventId.make("evt-open-user-input-2"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-open-user-input"),
+        occurredAt: "2026-02-26T13:00:01.000Z",
+        commandId: CommandId.make("cmd-open-user-input-2"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-open-user-input-2"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-open-user-input"),
+          projectId: ProjectId.make("project-open-user-input"),
+          title: "Thread Open User Input",
+          modelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          runtimeMode: "approval-required",
+          interactionMode: "default",
+          branch: null,
+          worktreePath: null,
+          createdAt: "2026-02-26T13:00:01.000Z",
+          updatedAt: "2026-02-26T13:00:01.000Z",
+        },
+      });
+
+      // Noise the summary must read past: the request it counts is a handful of
+      // rows among a thread's whole tool timeline.
+      yield* appendAndProject({
+        type: "thread.activity-appended",
+        eventId: EventId.make("evt-open-user-input-3"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-open-user-input"),
+        occurredAt: "2026-02-26T13:00:02.000Z",
+        commandId: CommandId.make("cmd-open-user-input-3"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-open-user-input-3"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-open-user-input"),
+          activity: {
+            id: EventId.make("activity-open-user-input-tool"),
+            tone: "tool",
+            kind: "tool.updated",
+            summary: "Tool progress",
+            payload: { requestId: "not-a-user-input-request" },
+            turnId: null,
+            createdAt: "2026-02-26T13:00:02.000Z",
+          },
+        },
+      });
+
+      yield* appendAndProject({
+        type: "thread.activity-appended",
+        eventId: EventId.make("evt-open-user-input-4"),
+        aggregateKind: "thread",
+        aggregateId: ThreadId.make("thread-open-user-input"),
+        occurredAt: "2026-02-26T13:00:03.000Z",
+        commandId: CommandId.make("cmd-open-user-input-4"),
+        causationEventId: null,
+        correlationId: CorrelationId.make("cmd-open-user-input-4"),
+        metadata: {},
+        payload: {
+          threadId: ThreadId.make("thread-open-user-input"),
+          activity: {
+            id: EventId.make("activity-open-user-input-requested"),
+            tone: "info",
+            kind: "user-input.requested",
+            summary: "User input requested",
+            payload: { requestId: "user-input-request-open-1" },
+            turnId: null,
+            createdAt: "2026-02-26T13:00:03.000Z",
+          },
+        },
+      });
+
+      const threadRows = yield* sql<{
+        readonly pendingUserInputCount: number;
+      }>`
+        SELECT pending_user_input_count AS "pendingUserInputCount"
+        FROM projection_threads
+        WHERE thread_id = 'thread-open-user-input'
+      `;
+      assert.deepEqual(threadRows, [{ pendingUserInputCount: 1 }]);
+    }),
+  );
+
   it.effect("ignores non-stale provider approval response failures", () =>
     Effect.gen(function* () {
       const projectionPipeline = yield* OrchestrationProjectionPipeline;

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

@@ -130,6 +130,17 @@ function isStalePendingApprovalFailureDetail(detail: string | null): boolean {
   );
 }
 
+/**
+ * The only activity kinds `derivePendingUserInputCountFromActivities` reacts to.
+ * Every other kind is skipped, so the query feeding it filters on these — keep
+ * the two in step.
+ */
+const PENDING_USER_INPUT_ACTIVITY_KINDS = [
+  "user-input.requested",
+  "user-input.resolved",
+  "provider.user-input.respond.failed",
+] as const;
+
 function derivePendingUserInputCountFromActivities(
   activities: ReadonlyArray<ProjectionThreadActivity>,
 ): number {
@@ -562,10 +573,18 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti
         return;
       }
 
+      // Only the user-input activity kinds, not the whole timeline: the single
+      // consumer below reduces them to one integer, while a thread's activity
+      // rows carry every tool payload it has produced. This runs on every event
+      // in the thread, so reading them in full makes each event cost the
+      // thread's entire history.
       const [messages, proposedPlans, activities, pendingApprovals] = yield* Effect.all([
         projectionThreadMessageRepository.listByThreadId({ threadId }),
         projectionThreadProposedPlanRepository.listByThreadId({ threadId }),
-        projectionThreadActivityRepository.listByThreadId({ threadId }),
+        projectionThreadActivityRepository.listByThreadIdAndKinds({
+          threadId,
+          kinds: PENDING_USER_INPUT_ACTIVITY_KINDS,
+        }),
         projectionPendingApprovalRepository.listByThreadId({ threadId }),
       ]);
 

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

@@ -0,0 +1,126 @@
+import { EventId, ThreadId } from "@t3tools/contracts";
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import { SqlitePersistenceMemory } from "./Sqlite.ts";
+import { ProjectionThreadActivityRepositoryLive } from "./ProjectionThreadActivities.ts";
+import { ProjectionThreadActivityRepository } from "../Services/ProjectionThreadActivities.ts";
+import type { ProjectionThreadActivity } from "../Services/ProjectionThreadActivities.ts";
+
+const threadId = ThreadId.make("thread-activity-kinds");
+const otherThreadId = ThreadId.make("thread-activity-kinds-other");
+
+const activity = (
+  id: string,
+  kind: string,
+  createdAt: string,
+  overrides: Partial<ProjectionThreadActivity> = {},
+): ProjectionThreadActivity => ({
+  activityId: EventId.make(id),
+  threadId,
+  turnId: null,
+  tone: "info",
+  kind,
+  summary: `${kind} ${id}`,
+  payload: { requestId: `request-${id}` },
+  createdAt,
+  ...overrides,
+});
+
+const layer = it.layer(
+  Layer.mergeAll(
+    ProjectionThreadActivityRepositoryLive.pipe(Layer.provideMerge(SqlitePersistenceMemory)),
+    SqlitePersistenceMemory,
+  ),
+);
+
+layer("ProjectionThreadActivityRepository.listByThreadIdAndKinds", (it) => {
+  const seed = Effect.fn("seed")(function* () {
+    const activities = yield* ProjectionThreadActivityRepository;
+    yield* activities.upsert(activity("a1", "user-input.requested", "2026-03-24T00:00:01.000Z"));
+    yield* activities.upsert(activity("a2", "tool.updated", "2026-03-24T00:00:02.000Z"));
+    yield* activities.upsert(activity("a3", "user-input.resolved", "2026-03-24T00:00:03.000Z"));
+    yield* activities.upsert(activity("a4", "tool.completed", "2026-03-24T00:00:04.000Z"));
+    yield* activities.upsert(
+      activity("a5", "user-input.requested", "2026-03-24T00:00:05.000Z", {
+        threadId: otherThreadId,
+      }),
+    );
+    return activities;
+  });
+
+  it.effect("returns only the requested kinds", () =>
+    Effect.gen(function* () {
+      const activities = yield* seed();
+
+      const rows = yield* activities.listByThreadIdAndKinds({
+        threadId,
+        kinds: ["user-input.requested", "user-input.resolved"],
+      });
+
+      assert.deepStrictEqual(
+        rows.map((row) => row.activityId),
+        ["a1", "a3"],
+      );
+    }),
+  );
+
+  it.effect("keeps the same ordering as the unfiltered list", () =>
+    Effect.gen(function* () {
+      const activities = yield* seed();
+
+      const all = yield* activities.listByThreadId({ threadId });
+      const filtered = yield* activities.listByThreadIdAndKinds({
+        threadId,
+        kinds: ["user-input.requested", "user-input.resolved", "tool.updated", "tool.completed"],
+      });
+
+      assert.deepStrictEqual(
+        filtered.map((row) => row.activityId),
+        all.map((row) => row.activityId),
+      );
+    }),
+  );
+
+  it.effect("decodes the payload the same way the unfiltered list does", () =>
+    Effect.gen(function* () {
+      const activities = yield* seed();
+
+      const [filtered] = yield* activities.listByThreadIdAndKinds({
+        threadId,
+        kinds: ["user-input.requested"],
+      });
+      const all = yield* activities.listByThreadId({ threadId });
+      const unfiltered = all.find((row) => row.activityId === "a1");
+
+      assert.deepStrictEqual(filtered, unfiltered);
+    }),
+  );
+
+  it.effect("does not reach across threads", () =>
+    Effect.gen(function* () {
+      const activities = yield* seed();
+
+      const rows = yield* activities.listByThreadIdAndKinds({
+        threadId: otherThreadId,
+        kinds: ["user-input.requested"],
+      });
+
+      assert.deepStrictEqual(
+        rows.map((row) => row.activityId),
+        ["a5"],
+      );
+    }),
+  );
+
+  it.effect("returns nothing for an empty kind list without querying", () =>
+    Effect.gen(function* () {
+      const activities = yield* seed();
+
+      const rows = yield* activities.listByThreadIdAndKinds({ threadId, kinds: [] });
+
+      assert.deepStrictEqual(rows, []);
+    }),
+  );
+});

apps/server/src/persistence/Layers/ProjectionThreadActivities.ts

@@ -10,6 +10,7 @@ import { toPersistenceDecodeError, toPersistenceSqlError } from "../Errors.ts";
 
 import {
   DeleteProjectionThreadActivitiesInput,
+  ListProjectionThreadActivitiesByKindInput,
   ListProjectionThreadActivitiesInput,
   ProjectionThreadActivity,
   ProjectionThreadActivityRepository,
@@ -30,6 +31,20 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st
       : toPersistenceSqlError(sqlOperation)(cause);
 }
 
+const toProjectionThreadActivity = (
+  row: typeof ProjectionThreadActivityDbRowSchema.Type,
+): ProjectionThreadActivity => ({
+  activityId: row.activityId,
+  threadId: row.threadId,
+  turnId: row.turnId,
+  tone: row.tone,
+  kind: row.kind,
+  summary: row.summary,
+  payload: row.payload,
+  ...(row.sequence !== null ? { sequence: row.sequence } : {}),
+  createdAt: row.createdAt,
+});
+
 const makeProjectionThreadActivityRepository = Effect.gen(function* () {
   const sql = yield* SqlClient.SqlClient;
 
@@ -97,6 +112,32 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
       `,
   });
 
+  const listProjectionThreadActivityRowsByKind = SqlSchema.findAll({
+    Request: ListProjectionThreadActivitiesByKindInput,
+    Result: ProjectionThreadActivityDbRowSchema,
+    execute: ({ threadId, kinds }) =>
+      sql`
+        SELECT
+          activity_id AS "activityId",
+          thread_id AS "threadId",
+          turn_id AS "turnId",
+          tone,
+          kind,
+          summary,
+          payload_json AS "payload",
+          sequence,
+          created_at AS "createdAt"
+        FROM projection_thread_activities
+        WHERE thread_id = ${threadId}
+          AND kind IN ${sql.in(kinds)}
+        ORDER BY
+          CASE WHEN sequence IS NULL THEN 0 ELSE 1 END ASC,
+          sequence ASC,
+          created_at ASC,
+          activity_id ASC
+      `,
+  });
+
   const deleteProjectionThreadActivityRows = SqlSchema.void({
     Request: DeleteProjectionThreadActivitiesInput,
     execute: ({ threadId }) =>
@@ -124,21 +165,23 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
           "ProjectionThreadActivityRepository.listByThreadId:decodeRows",
         ),
       ),
-      Effect.map((rows) =>
-        rows.map((row) => ({
-          activityId: row.activityId,
-          threadId: row.threadId,
-          turnId: row.turnId,
-          tone: row.tone,
-          kind: row.kind,
-          summary: row.summary,
-          payload: row.payload,
-          ...(row.sequence !== null ? { sequence: row.sequence } : {}),
-          createdAt: row.createdAt,
-        })),
-      ),
+      Effect.map((rows) => rows.map(toProjectionThreadActivity)),
     );
 
+  const listByThreadIdAndKinds: ProjectionThreadActivityRepositoryShape["listByThreadIdAndKinds"] =
+    (input) =>
+      input.kinds.length === 0
+        ? Effect.succeed([])
+        : listProjectionThreadActivityRowsByKind(input).pipe(
+            Effect.mapError(
+              toPersistenceSqlOrDecodeError(
+                "ProjectionThreadActivityRepository.listByThreadIdAndKinds:query",
+                "ProjectionThreadActivityRepository.listByThreadIdAndKinds:decodeRows",
+              ),
+            ),
+            Effect.map((rows) => rows.map(toProjectionThreadActivity)),
+          );
+
   const deleteByThreadId: ProjectionThreadActivityRepositoryShape["deleteByThreadId"] = (input) =>
     deleteProjectionThreadActivityRows(input).pipe(
       Effect.mapError(
@@ -149,6 +192,7 @@ const makeProjectionThreadActivityRepository = Effect.gen(function* () {
   return {
     upsert,
     listByThreadId,
+    listByThreadIdAndKinds,
     deleteByThreadId,
   } satisfies ProjectionThreadActivityRepositoryShape;
 });