perf(server): port #13765 to V2 — background PR discovery reads only threads that can still settle

Main's periodic discovery and settlement sweeps stopped reading settled
threads. V2's settlement sweep already reads candidates only
(getSettlementCandidates filters settled, pinned, archived, and busy
threads in SQL), but V2's PR discovery ran a full getShellSnapshot every
minute: every thread, active and archived, settled or not, with every
thread's run, item, and session subqueries. Discovery then dropped
archived and settled threads unless they were in backfill.

getShellSnapshot takes an `unsettledOnly` option (SQL and memory store),
and the periodic discovery sweep reads active, unsettled threads unless a
backfill pass is pending, as on main. A pending backfill entry for a
thread with no branch is cleared so it cannot keep every pass on the
full read.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: juliusmarminge

apps/server/src/orchestration-v2/Orchestrator.ts

@@ -268,6 +268,8 @@ export interface OrchestratorV2Shape {
   >;
   readonly getShellSnapshot: (options?: {
     readonly location?: "active" | "archive";
+    /** Background sweeps only: skips settled threads. */
+    readonly unsettledOnly?: boolean;
   }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, OrchestratorV2Error>;
   readonly getThreadShell: (
     threadId: ThreadId,

apps/server/src/orchestration-v2/ProjectionSettlement.test.ts

@@ -344,6 +344,31 @@ for (const [name, testLayer] of [
   );
 }
 
+for (const [name, testLayer] of [
+  ["sql", SqlLayer],
+  ["memory", layerMemory],
+] as const) {
+  it.effect(`${name}: an unsettled-only shell read skips settled threads`, () =>
+    Effect.gen(function* () {
+      const store = yield* ProjectionStoreV2;
+      const open = yield* createThread("unsettled-open");
+      const reopened = yield* createThread("unsettled-reopened", { settledOverride: "active" });
+      yield* createThread("unsettled-manual", { settledOverride: "settled", settledAt: old });
+      yield* createThread("unsettled-auto", { settledAt: old });
+      yield* createThread("unsettled-archived", { archivedAt: old });
+
+      const shell = yield* store.getShellSnapshot({ location: "active", unsettledOnly: true });
+      assert.deepEqual(
+        new Set(shell.threads.map((thread) => thread.id)),
+        new Set([open, reopened]),
+      );
+      assert.equal(shell.archivedThreads.length, 0);
+      const all = yield* store.getShellSnapshot({ location: "active" });
+      assert.equal(all.threads.length, 4);
+    }).pipe(Effect.provide(testLayer)),
+  );
+}
+
 it.effect(
   "reads settlement candidates and thread metadata without loading historical or archived payloads",
   () =>

apps/server/src/orchestration-v2/ProjectionStore.ts

@@ -319,8 +319,13 @@ export interface ProjectionStoreV2Shape {
   readonly apply: (
     event: OrchestrationV2DomainEvent,
   ) => Effect.Effect<void, ProjectionStoreV2Error>;
+  /**
+   * `unsettledOnly` is for background sweeps, not clients: it skips settled
+   * threads before any of their run, item or session rows are read.
+   */
   readonly getShellSnapshot: (options?: {
     readonly location?: "active" | "archive";
+    readonly unsettledOnly?: boolean;
   }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, ProjectionStoreV2Error>;
   readonly getThreadShell: (
     threadId: ThreadId,
@@ -4677,7 +4682,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           ),
         );
 
-    const selectShellThreadRows = (threadId?: ThreadId, location?: "active" | "archive") =>
+    const selectShellThreadRows = (
+      threadId?: ThreadId,
+      location?: "active" | "archive",
+      unsettledOnly = false,
+    ) =>
       sql<ShellThreadRow>`
             SELECT
               t.thread_id,
@@ -4815,6 +4824,10 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
                 : location === "archive"
                   ? sql` AND json_extract(t.payload_json, '$.archivedAt') IS NOT NULL`
                   : sql``
+            }${
+              unsettledOnly
+                ? sql` AND json_extract(t.payload_json, '$.settledAt') IS NULL AND json_extract(t.payload_json, '$.settledOverride') IS NOT 'settled'`
+                : sql``
             }
             ORDER BY t.updated_at ASC, t.thread_id ASC
           `;
@@ -5169,7 +5182,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
       sql
         .withTransaction(
           Effect.gen(function* () {
-            const targetThreadRows = yield* selectShellThreadRows(undefined, options?.location);
+            const targetThreadRows = yield* selectShellThreadRows(
+              undefined,
+              options?.location,
+              options?.unsettledOnly ?? false,
+            );
             const targetThreadIds = new Set(
               targetThreadRows.map((row) => ThreadId.make(row.thread_id)),
             );
@@ -5393,6 +5410,13 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
           const existing = (yield* Ref.get(replayState)).projections;
           const selectedThreadIds = [...existing.entries()]
             .filter(([, projection]) => {
+              if (
+                options?.unsettledOnly &&
+                (projection.thread.settledAt !== null ||
+                  projection.thread.settledOverride === "settled")
+              ) {
+                return false;
+              }
               if (options?.location === "active") return projection.thread.archivedAt === null;
               if (options?.location === "archive") return projection.thread.archivedAt !== null;
               return true;

apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts

@@ -18,6 +18,7 @@ import * as Option from "effect/Option";
 import * as PubSub from "effect/PubSub";
 import * as Queue from "effect/Queue";
 import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
 
 import { GitManager } from "../git/GitManager.ts";
 import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
@@ -164,13 +165,15 @@ describe("ThreadPullRequestServiceV2 reads", () => {
         const other = threadShell("other-thread");
         const activation = yield* Deferred.make<void>();
         const events = yield* PubSub.unbounded<OrchestrationV2DomainEvent>();
-        // Each read: the thread id for a one-thread read, null for a full read.
-        const reads = yield* Queue.unbounded<ThreadId | null>();
+        // Each read: the thread id for a one-thread read, or the full read's options.
+        const reads = yield* Queue.unbounded<
+          ThreadId | { readonly location?: string; readonly unsettledOnly?: boolean }
+        >();
         const dependencies = Layer.mergeAll(
           Layer.mock(OrchestratorV2)({
             streamDomainEvents: Stream.fromPubSub(events),
-            getShellSnapshot: () =>
-              Queue.offer(reads, null).pipe(
+            getShellSnapshot: (options) =>
+              Queue.offer(reads, options ?? {}).pipe(
                 Effect.as({
                   schemaVersion: 2,
                   snapshotSequence: 1,
@@ -205,8 +208,8 @@ describe("ThreadPullRequestServiceV2 reads", () => {
           const service = yield* make;
           yield* service.start();
           yield* Deferred.succeed(activation, undefined);
-          // Startup backfill is a full read.
-          expect(yield* Queue.take(reads)).toBeNull();
+          // Startup backfill reads every active thread.
+          expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: false });
           yield* service.drain;
           yield* PubSub.publish(events, {
             type: "thread.metadata-updated",
@@ -244,4 +247,65 @@ describe("ThreadPullRequestServiceV2 reads", () => {
       }),
     ),
   );
+
+  it.effect("periodic sweeps read only unsettled threads once backfill is done", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const settled = {
+          ...threadShell("settled-thread"),
+          branch: "feature/settled",
+          settledOverride: "settled" as const,
+          settledAt: NOW,
+        };
+        const activation = yield* Deferred.make<void>();
+        const reads = yield* Queue.unbounded<{
+          readonly location?: string;
+          readonly unsettledOnly?: boolean;
+        }>();
+        const dependencies = Layer.mergeAll(
+          Layer.mock(OrchestratorV2)({
+            streamDomainEvents: Stream.never,
+            getShellSnapshot: (options) =>
+              Queue.offer(reads, options ?? {}).pipe(
+                Effect.as({
+                  schemaVersion: 2,
+                  snapshotSequence: 1,
+                  // The fake honors unsettledOnly like the store does.
+                  threads: options?.unsettledOnly ? [] : [settled],
+                  archivedThreads: [],
+                }),
+              ),
+          }),
+          Layer.mock(ProjectionSnapshotQuery)({
+            getProjectShellsWithoutEnrichment: () => Effect.succeed([]),
+          }),
+          Layer.mock(GitManager)({}),
+          Layer.mock(PullRequestService)({}),
+          Layer.mock(RepositoryIdentityResolver)({}),
+          Layer.succeed(ServerActivation, Deferred.await(activation)),
+          Layer.succeed(
+            Crypto.Crypto,
+            Crypto.make({
+              randomBytes: (size) => new Uint8Array(size).fill(1),
+              digest: (_algorithm, data) => Effect.succeed(data),
+            }),
+          ),
+          FileSystem.layerNoop({}),
+        );
+
+        yield* Effect.gen(function* () {
+          const service = yield* make;
+          yield* service.start();
+          yield* Deferred.succeed(activation, undefined);
+          // Backfill finds the settled branch thread and must read it again.
+          expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: false });
+          yield* service.drain;
+          // Its project is gone, so backfill finishes it on the first pass.
+          yield* TestClock.adjust("1 minute");
+          expect(yield* Queue.take(reads)).toEqual({ location: "active", unsettledOnly: true });
+          yield* service.drain;
+        }).pipe(Effect.provide(dependencies));
+      }),
+    ),
+  );
 });

apps/server/src/orchestration-v2/ThreadPullRequestService.ts

@@ -121,11 +121,16 @@ export const make = Effect.gen(function* () {
 
   /**
    * A sweep for one thread reads only that thread's shell, not every thread's.
-   * Finished runs and checkpoints queue one of these each.
+   * Finished runs and checkpoints queue one of these each. A sweep over all
+   * threads reads only active, unsettled ones, since discovery skips the rest;
+   * backfill looks up settled threads, so its passes read every active thread.
    */
-  const readThreadSnapshot = (threadId: ThreadId | null) =>
+  const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) =>
     threadId === null
-      ? orchestrator.getShellSnapshot()
+      ? orchestrator.getShellSnapshot({
+          location: "active",
+          unsettledOnly: !(backfill || pendingBackfill.size > 0),
+        })
       : Effect.gen(function* () {
           // Read the sequence first. The thread is then at least this new, so a
           // sync guarded by the sequence is rejected rather than missing a change.
@@ -138,7 +143,7 @@ export const make = Effect.gen(function* () {
     request: RefreshRequest,
   ) {
     const [threadSnapshot, projectShells] = yield* Effect.all([
-      readThreadSnapshot(request.threadId),
+      readThreadSnapshot(request),
       snapshots.getProjectShellsWithoutEnrichment(),
     ]);
     const projects = new Map(projectShells.map((project) => [project.id, project]));
@@ -152,8 +157,12 @@ export const make = Effect.gen(function* () {
         }
       }
     }
-    // A single-thread read only shows whether its own thread is gone.
-    const visibleThreadIds = new Set(threadSnapshot.threads.map((thread) => thread.id));
+    // A single-thread read only shows whether its own thread is gone. A thread
+    // with no branch has nothing to look up, and its entry would keep every
+    // periodic pass on the full read.
+    const visibleThreadIds = new Set(
+      threadSnapshot.threads.filter((thread) => thread.branch !== null).map((thread) => thread.id),
+    );
     const checkedIds = request.threadId === null ? [...pendingBackfill.keys()] : [request.threadId];
     for (const threadId of checkedIds) {
       if (!visibleThreadIds.has(threadId)) pendingBackfill.delete(threadId);