perf(server): port #13691 to V2 — per-thread settlement and PR checks no longer read every thread

Main made single-thread settlement and pull request sweeps read only that
thread. V2 had the same shape: every finished run, detached session,
checkpoint or metadata change queued a one-thread request, and each one
read every thread before filtering to the one it wanted.

- ThreadSettlementServiceV2: getSettlementCandidates takes an optional
  thread id and filters in SQL (and in the memory store), and a sweep
  with no candidates stops before reading projects.
- ThreadPullRequestServiceV2: a one-thread request reads the thread's
  sequence, then its shell via getThreadShell, instead of
  getShellSnapshot (every active and archived thread). As on main, a
  one-thread read only clears that thread's pending backfill.

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

Author: juliusmarminge

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

@@ -333,10 +333,10 @@ export interface ProjectionStoreV2Shape {
     readonly autoResume: boolean;
     readonly snooze: boolean;
   }) => Effect.Effect<ReadonlyArray<ProjectionLimitRecoveryCandidate>, ProjectionStoreV2Error>;
-  readonly getSettlementCandidates: () => Effect.Effect<
-    ReadonlyArray<ProjectionSettlementCandidate>,
-    ProjectionStoreV2Error
-  >;
+  /** Every candidate, or only `threadId` when a sweep checks one thread. */
+  readonly getSettlementCandidates: (
+    threadId?: ThreadId,
+  ) => Effect.Effect<ReadonlyArray<ProjectionSettlementCandidate>, ProjectionStoreV2Error>;
   /**
    * Active (not deleted, not archived) threads with at least one pull request
    * link, in shell snapshot order. Skips run, message and item reads.
@@ -4941,7 +4941,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
         return { providerThreadsByThreadId, pendingTurnItemsByThreadId };
       });
 
-    const getSettlementCandidates: ProjectionStoreV2Shape["getSettlementCandidates"] = () =>
+    const getSettlementCandidates: ProjectionStoreV2Shape["getSettlementCandidates"] = (threadId) =>
       sql
         .withTransaction(
           Effect.gen(function* () {
@@ -4969,7 +4969,7 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
               ORDER BY latest.ordinal DESC, latest.run_id DESC
               LIMIT 1
             )
-            WHERE t.deleted_at IS NULL
+            WHERE t.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}
               AND json_extract(t.payload_json, '$.archivedAt') IS NULL
               AND json_extract(t.payload_json, '$.settledOverride') IS NULL
               AND json_extract(t.payload_json, '$.pinnedAt') IS NULL
@@ -5430,12 +5430,13 @@ export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
           }
           return projection.thread;
         }),
-      getSettlementCandidates: () =>
+      getSettlementCandidates: (threadId) =>
         Effect.gen(function* () {
           const projections = (yield* Ref.get(replayState)).projections;
           return [...projections.values()]
             .filter(
               ({ thread, runs, runtimeRequests }) =>
+                (threadId === undefined || thread.id === threadId) &&
                 thread.deletedAt === null &&
                 thread.archivedAt === null &&
                 thread.settledOverride === null &&

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

@@ -1,9 +1,32 @@
-import { ProjectId, type OrchestrationProjectShell } from "@t3tools/contracts";
+import {
+  EventId,
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationProjectShell,
+  type OrchestrationV2DomainEvent,
+  type OrchestrationV2ThreadShell,
+} from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
+import * as Crypto from "effect/Crypto";
+import * as DateTime from "effect/DateTime";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Layer from "effect/Layer";
 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 { GitManager } from "../git/GitManager.ts";
+import { ProjectionSnapshotQuery } from "../orchestration/Services/ProjectionSnapshotQuery.ts";
+import { RepositoryIdentityResolver } from "../project/RepositoryIdentityResolver.ts";
+import { PullRequestService } from "../pullRequest/PullRequestService.ts";
+import { ServerActivation } from "../serverActivation.ts";
+import { OrchestratorV2 } from "./Orchestrator.ts";
 import {
+  make,
   projectWorkspaceMatchesSnapshot,
   resolveProjectForPullRequestDiscovery,
 } from "./ThreadPullRequestService.ts";
@@ -70,3 +93,130 @@ describe("ThreadPullRequestServiceV2 project guard", () => {
     expect(projectWorkspaceMatchesSnapshot(currentProject, "/workspace/original")).toBe(true);
   });
 });
+
+describe("ThreadPullRequestServiceV2 reads", () => {
+  const NOW = DateTime.makeUnsafe("2026-09-20T00:00:00.000Z");
+  const threadShell = (id: string): OrchestrationV2ThreadShell => {
+    const threadId = ThreadId.make(id);
+    return {
+      id: threadId,
+      projectId: ProjectId.make("project-1"),
+      title: id,
+      providerInstanceId: ProviderInstanceId.make("codex"),
+      modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      branch: null,
+      worktreePath: null,
+      activeProviderThreadId: null,
+      lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null },
+      forkedFrom: null,
+      createdBy: "user",
+      creationSource: "web",
+      activeRunId: null,
+      latestRunId: null,
+      status: "idle",
+      pendingRuntimeRequest: null,
+      latestVisibleMessage: null,
+      latestUserMessageAt: null,
+      hasActionableProposedPlan: false,
+      itemCount: 0,
+      visibleItemCount: 0,
+      createdAt: NOW,
+      updatedAt: NOW,
+      archivedAt: null,
+      settledOverride: null,
+      settledAt: null,
+      lastVisitedAt: null,
+      deletedAt: null,
+    };
+  };
+
+  it.effect("an event for one thread reads that thread's shell, not every thread's", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const thread = threadShell("updated-thread");
+        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>();
+        const dependencies = Layer.mergeAll(
+          Layer.mock(OrchestratorV2)({
+            streamDomainEvents: Stream.fromPubSub(events),
+            getShellSnapshot: () =>
+              Queue.offer(reads, null).pipe(
+                Effect.as({
+                  schemaVersion: 2,
+                  snapshotSequence: 1,
+                  threads: [thread, other],
+                  archivedThreads: [],
+                }),
+              ),
+            getThreadEventSequence: () => Effect.succeed(1),
+            getThreadShell: (threadId) =>
+              Queue.offer(reads, threadId).pipe(
+                Effect.as([thread, other].find((candidate) => candidate.id === threadId) ?? null),
+              ),
+          }),
+          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);
+          // Startup backfill is a full read.
+          expect(yield* Queue.take(reads)).toBeNull();
+          yield* service.drain;
+          yield* PubSub.publish(events, {
+            type: "thread.metadata-updated",
+            id: EventId.make("event:metadata"),
+            threadId: thread.id,
+            occurredAt: NOW,
+            payload: {
+              createdBy: thread.createdBy,
+              creationSource: thread.creationSource,
+              id: thread.id,
+              projectId: thread.projectId,
+              title: thread.title,
+              providerInstanceId: thread.providerInstanceId,
+              modelSelection: thread.modelSelection,
+              runtimeMode: thread.runtimeMode,
+              interactionMode: thread.interactionMode,
+              branch: thread.branch,
+              worktreePath: thread.worktreePath,
+              activeProviderThreadId: thread.activeProviderThreadId,
+              lineage: thread.lineage,
+              forkedFrom: null,
+              createdAt: thread.createdAt,
+              updatedAt: thread.updatedAt,
+              archivedAt: null,
+              settledOverride: null,
+              settledAt: null,
+              lastVisitedAt: null,
+              deletedAt: null,
+            },
+          });
+          expect(yield* Queue.take(reads)).toBe(thread.id);
+          yield* service.drain;
+          expect(yield* Queue.size(reads)).toBe(0);
+        }).pipe(Effect.provide(dependencies));
+      }),
+    ),
+  );
+});

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

@@ -114,11 +114,26 @@ 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.
+   */
+  const readThreadSnapshot = (threadId: ThreadId | null) =>
+    threadId === null
+      ? orchestrator.getShellSnapshot()
+      : 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.
+          const snapshotSequence = yield* orchestrator.getThreadEventSequence(threadId);
+          const thread = yield* orchestrator.getThreadShell(threadId);
+          return { snapshotSequence, threads: thread === null ? [] : [thread] };
+        });
+
   const synchronize = Effect.fn("ThreadPullRequestServiceV2.synchronize")(function* (
     request: RefreshRequest,
   ) {
     const [threadSnapshot, projectShells] = yield* Effect.all([
-      orchestrator.getShellSnapshot(),
+      readThreadSnapshot(request.threadId),
       snapshots.getProjectShellsWithoutEnrichment(),
     ]);
     const projects = new Map(projectShells.map((project) => [project.id, project]));
@@ -132,14 +147,15 @@ 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));
-    for (const threadId of pendingBackfill.keys()) {
+    const checkedIds = request.threadId === null ? [...pendingBackfill.keys()] : [request.threadId];
+    for (const threadId of checkedIds) {
       if (!visibleThreadIds.has(threadId)) pendingBackfill.delete(threadId);
     }
     const threads = threadSnapshot.threads.filter(
       (thread) =>
         thread.archivedAt === null &&
-        (request.threadId === null || thread.id === request.threadId) &&
         ((thread.settledOverride !== "settled" && thread.settledAt === null) ||
           request.threadId !== null ||
           pendingBackfill.has(thread.id)) &&

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

@@ -5,6 +5,7 @@ import {
   ProjectId,
   EventId,
   ProviderInstanceId,
+  ProviderSessionId,
   ThreadId,
   type OrchestrationProjectShell,
   type OrchestrationV2AppThread,
@@ -423,6 +424,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
   const snapshots = yield* Ref.make(options.snapshot);
   const snapshotReadCount = yield* Ref.make(0);
   const snapshotReads = yield* Queue.unbounded<number>();
+  const candidateReads = yield* Ref.make<ReadonlyArray<string | undefined>>([]);
   const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS);
   const settingsChanges = yield* PubSub.unbounded<ServerSettings>();
   const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>();
@@ -501,11 +503,16 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
         Ref.get(snapshots).pipe(Effect.map((snapshot) => snapshot.projects)),
     }),
     Layer.mock(ProjectionStoreV2)({
-      getSettlementCandidates: () =>
-        Ref.updateAndGet(snapshotReadCount, (count) => count + 1).pipe(
+      getSettlementCandidates: (threadId) =>
+        Ref.update(candidateReads, (reads) => [...reads, threadId]).pipe(
+          Effect.andThen(Ref.updateAndGet(snapshotReadCount, (count) => count + 1)),
           Effect.tap((count) => Queue.offer(snapshotReads, count)),
           Effect.andThen(Ref.get(snapshots)),
-          Effect.map((snapshot) => snapshot.threads),
+          Effect.map((snapshot) =>
+            threadId === undefined
+              ? snapshot.threads
+              : snapshot.threads.filter((thread) => thread.id === threadId),
+          ),
         ),
       getThread: (threadId) => {
         const thread = options.currentThreads?.find((candidate) => candidate.id === threadId);
@@ -544,6 +551,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
     snapshots,
     snapshotReadCount,
     snapshotReads,
+    candidateReads,
     commands,
     branchCalls,
     summaryCalls,
@@ -1006,3 +1014,38 @@ describe("ThreadSettlementServiceV2 terminals", () => {
     ),
   );
 });
+
+describe("ThreadSettlementServiceV2 single-thread sweeps", () => {
+  it.effect("a finished run reads only its own thread's settlement candidate", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        yield* TestClock.setTime(Date.parse(NOW));
+        const finished = makeThread("finished-run");
+        const other = makeThread("other-thread");
+        const fixture = yield* makeHarness({
+          snapshot: makeSnapshot([finished, other]),
+          settings: { ...DEFAULT_SERVER_SETTINGS, sidebarAutoSettleAfterDays: 3 },
+        });
+
+        yield* Effect.gen(function* () {
+          const service = yield* ThreadSettlementService.ThreadSettlementServiceV2;
+          yield* startHarness(service, fixture.activation, fixture.snapshotReads);
+          yield* Ref.set(fixture.candidateReads, []);
+          yield* fixture.publishEvent({
+            type: "provider-session.detached",
+            id: EventId.make("event:detached"),
+            threadId: finished.id,
+            occurredAt: DateTime.makeUnsafe(NOW),
+            payload: {
+              providerSessionId: ProviderSessionId.make("provider-session:finished"),
+              detachedAt: DateTime.makeUnsafe(NOW),
+            },
+          });
+          yield* Queue.take(fixture.snapshotReads);
+          yield* service.drain;
+          assert.deepStrictEqual(yield* Ref.get(fixture.candidateReads), [finished.id]);
+        }).pipe(Effect.provide(fixture.layer));
+      }),
+    ),
+  );
+});

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

@@ -268,19 +268,17 @@ export const make = Effect.gen(function* () {
     if (!autoSettlementConfigured(settings)) {
       return;
     }
-    const threads = yield* projections.getSettlementCandidates();
+    // A sweep for one thread reads only that thread's candidate row.
+    const threads = yield* projections.getSettlementCandidates(threadId);
+    if (threads.length === 0) return;
     const projectShells = yield* snapshots.getProjectShellsWithoutEnrichment();
     const nowMs = DateTime.toEpochMillis(yield* DateTime.now);
     const projects = new Map(projectShells.map((project) => [project.id, project]));
     // A merge event re-sweeps every candidate, not just the threads linked to
     // the merged pull request: most threads carry no link and settle from
     // their branch lookup, which would otherwise wait for the next minute's
     // sweep on a possibly stale cached answer.
-    const candidates = threads.filter(
-      (thread) =>
-        (threadId === undefined || thread.id === threadId) &&
-        isAutoSettlementCandidate(thread, nowMs),
-    );
+    const candidates = threads.filter((thread) => isAutoSettlementCandidate(thread, nowMs));
 
     const settleThread = Effect.fn("ThreadSettlementServiceV2.settleThread")(
       function* (thread: (typeof candidates)[number], pullRequest: SettlementPullRequest | null) {