fix(server): promote the pull-request sweep and plan-flag freeze fix to bkt3

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

@@ -2716,18 +2716,31 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
           return yield* new ProjectionStoreThreadNotFoundError({ threadId });
         }
 
+        // T3-CUSTOM(expbkt3): a run filter reads only that run's items through the
+        // (thread_id, run_id) index. In ordinal order SQLite walked every item of the
+        // thread instead, which took 0.3-2.6 s on a cold 30,000-item thread and froze
+        // the server (agent toolkit plan-flag checks run it for up to 40 sessions).
+        const turnItemRunIdsWithoutNull =
+          filter?.turnItemRunIds === undefined || filter.turnItemRunIds.includes(null)
+            ? undefined
+            : filter.turnItemRunIds.filter((id): id is RunId => id !== null);
+        const turnItemRunIndex =
+          filter?.turnItemRunId !== undefined || turnItemRunIdsWithoutNull !== undefined
+            ? sql`INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx`
+            : sql``;
+
         const boundedTurnItemRows =
           fields !== undefined && !fields.includes("turnItems")
             ? []
             : window === undefined
               ? yield* sql<PayloadRow>`
                 SELECT payload_json
-                FROM orchestration_v2_projection_turn_items
+                FROM orchestration_v2_projection_turn_items ${turnItemRunIndex} /* T3-CUSTOM(expbkt3): run filters use the run index. */
                 WHERE thread_id = ${threadId}
                   ${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`}
                   ${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`}
                   ${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`}
-                  ${filter?.turnItemRunIds === undefined ? sql`` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`}
+                  ${filter?.turnItemRunIds === undefined ? sql`` : turnItemRunIdsWithoutNull !== undefined ? sql`AND run_id IN (SELECT value FROM json_each(${encodeIdList(turnItemRunIdsWithoutNull)}))` : sql`AND (run_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemRunIds.filter((id): id is RunId => id !== null))})) OR (${filter.turnItemRunIds.includes(null) ? 1 : 0} = 1 AND run_id IS NULL))`} /* T3-CUSTOM(expbkt3): an OR here defeats the run index. */
                 ORDER BY ordinal ASC, turn_item_id ASC
               `
               : yield* sql<PayloadRow>`

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

@@ -12,6 +12,8 @@ import {
 } from "@t3tools/contracts";
 import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
 import * as Cause from "effect/Cause";
+// T3-CUSTOM(expbkt3): the idle-thread rotation reads the clock.
+import * as Clock from "effect/Clock";
 import * as Context from "effect/Context";
 import * as Crypto from "effect/Crypto";
 import * as Effect from "effect/Effect";
@@ -28,6 +30,8 @@ import * as PullRequestService from "../pullRequest/PullRequestService.ts";
 import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
 import { forkParked } from "../serverActivation.ts";
 import * as Orchestrator from "./Orchestrator.ts";
+// T3-CUSTOM(expbkt3): idle threads ride a slow rotation in the periodic sweep.
+import { includeInPeriodicSweep } from "./threadPullRequestSweep.expbkt3.ts";
 
 class ThreadPullRequestServiceV2 extends Context.Service<
   ThreadPullRequestServiceV2,
@@ -106,6 +110,8 @@ export const make = Effect.gen(function* () {
   const crypto = yield* Crypto.Crypto;
   const fileSystem = yield* FileSystem.FileSystem;
   const pendingBackfill = new Map<ThreadId, number>();
+  // T3-CUSTOM(expbkt3): counts all-thread passes for the idle-thread rotation.
+  let periodicPass = 0;
 
   const finishBackfill = (threads: ReadonlyArray<Pick<OrchestrationV2ThreadShell, "id">>) => {
     for (const thread of threads) pendingBackfill.delete(thread.id);
@@ -167,8 +173,17 @@ export const make = Effect.gen(function* () {
     for (const threadId of checkedIds) {
       if (!visibleThreadIds.has(threadId)) pendingBackfill.delete(threadId);
     }
+    // T3-CUSTOM(expbkt3): BEGIN — an all-thread pass checks idle threads on a rotation.
+    const nowMs = yield* Clock.currentTimeMillis;
+    const pass = periodicPass;
+    if (request.threadId === null) periodicPass += 1;
+    // T3-CUSTOM(expbkt3): END
     const threads = threadSnapshot.threads.filter(
       (thread) =>
+        // T3-CUSTOM(expbkt3): a one-thread check and backfill are never rotated.
+        (request.threadId !== null ||
+          pendingBackfill.has(thread.id) ||
+          includeInPeriodicSweep(thread, nowMs, pass)) &&
         thread.archivedAt === null &&
         ((thread.settledOverride !== "settled" && thread.settledAt === null) ||
           request.threadId !== null ||

apps/server/src/orchestration-v2/threadPullRequestSweep.expbkt3.test.ts

@@ -0,0 +1,196 @@
+import {
+  ProjectId,
+  ProviderInstanceId,
+  ThreadId,
+  type OrchestrationProjectShell,
+  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 Queue from "effect/Queue";
+import * as Stream from "effect/Stream";
+import * as TestClock from "effect/testing/TestClock";
+
+import * as GitManager from "../git/GitManager.ts";
+import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
+import * as PullRequestService from "../pullRequest/PullRequestService.ts";
+import * as ServerActivation from "../serverActivation.ts";
+import * as Orchestrator from "./Orchestrator.ts";
+import * as ProjectStore from "./ProjectStore.ts";
+import * as ThreadPullRequestService from "./ThreadPullRequestService.ts";
+import {
+  IDLE_SWEEP_EVERY_PASSES,
+  idleSweepSlot,
+  includeInPeriodicSweep,
+  isIdleForSweep,
+} from "./threadPullRequestSweep.expbkt3.ts";
+
+const NOW = DateTime.makeUnsafe("2026-10-19T10:00:00.000Z");
+const NOW_MS = DateTime.toEpochMillis(NOW);
+const hoursAgo = (hours: number) => DateTime.makeUnsafe(NOW_MS - hours * 60 * 60 * 1000);
+
+const threadShell = (
+  id: string,
+  overrides: Partial<OrchestrationV2ThreadShell> = {},
+): OrchestrationV2ThreadShell => {
+  const threadId = ThreadId.make(id);
+  return {
+    id: threadId,
+    projectId: ProjectId.make(`project-${id}`),
+    title: id,
+    providerInstanceId: ProviderInstanceId.make("codex"),
+    modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
+    runtimeMode: "full-access",
+    interactionMode: "default",
+    branch: `feature/${id}`,
+    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: hoursAgo(24 * 30),
+    updatedAt: NOW,
+    archivedAt: null,
+    settledOverride: null,
+    settledAt: null,
+    lastVisitedAt: null,
+    deletedAt: null,
+    ...overrides,
+  };
+};
+
+describe("periodic pull-request sweep rotation", () => {
+  it("treats a thread as idle only after a day with no run, message or live work", () => {
+    expect(isIdleForSweep(threadShell("old", { latestRunCompletedAt: hoursAgo(25) }), NOW_MS)).toBe(
+      true,
+    );
+    expect(
+      isIdleForSweep(threadShell("recent", { latestRunCompletedAt: hoursAgo(2) }), NOW_MS),
+    ).toBe(false);
+    expect(
+      isIdleForSweep(threadShell("messaged", { latestUserMessageAt: hoursAgo(1) }), NOW_MS),
+    ).toBe(false);
+    expect(isIdleForSweep(threadShell("new", { createdAt: hoursAgo(3) }), NOW_MS)).toBe(false);
+    expect(
+      isIdleForSweep(
+        threadShell("waiting", {
+          latestRunCompletedAt: hoursAgo(48),
+          activityRunStatus: "waiting",
+        }),
+        NOW_MS,
+      ),
+    ).toBe(false);
+  });
+
+  it("checks every idle thread exactly once per rotation and active threads on every pass", () => {
+    const idle = Array.from({ length: 200 }, (_, index) =>
+      threadShell(`idle-${index}`, { latestRunCompletedAt: hoursAgo(72) }),
+    );
+    const active = threadShell("active", { latestRunCompletedAt: hoursAgo(1) });
+    const checks = new Map<string, number>();
+    for (let pass = 0; pass < IDLE_SWEEP_EVERY_PASSES; pass += 1) {
+      for (const thread of [...idle, active]) {
+        if (includeInPeriodicSweep(thread, NOW_MS, pass)) {
+          checks.set(thread.id, (checks.get(thread.id) ?? 0) + 1);
+        }
+      }
+    }
+    expect(idle.every((thread) => checks.get(thread.id) === 1)).toBe(true);
+    expect(checks.get(active.id)).toBe(IDLE_SWEEP_EVERY_PASSES);
+    // The slots spread the idle threads instead of bunching them on one pass.
+    const perSlot = new Set(idle.map((thread) => idleSweepSlot(thread.id)));
+    expect(perSlot.size).toBeGreaterThan(IDLE_SWEEP_EVERY_PASSES / 2);
+  });
+
+  it.effect(
+    "the service looks up an idle thread once per rotation and an active one every pass",
+    () =>
+      Effect.scoped(
+        Effect.gen(function* () {
+          const idle = threadShell("idle", { latestRunCompletedAt: hoursAgo(72) });
+          const active = threadShell("active", { latestRunCompletedAt: hoursAgo(1) });
+          const project = (thread: OrchestrationV2ThreadShell): OrchestrationProjectShell => ({
+            id: thread.projectId,
+            title: thread.title,
+            workspaceRoot: `/workspace/${thread.id}`,
+            defaultModelSelection: null,
+            scripts: [],
+            repositoryIdentity: null,
+            createdAt: "2026-08-01T00:00:00.000Z",
+            updatedAt: "2026-08-01T00:00:00.000Z",
+          });
+          const activation = yield* Deferred.make<void>();
+          const passes = yield* Queue.unbounded<void>();
+          const lookups = new Map<string, number>();
+          const layerDependencies = Layer.mergeAll(
+            Layer.mock(Orchestrator.OrchestratorV2)({
+              streamDomainEvents: Stream.never,
+              getShellSnapshot: () =>
+                Queue.offer(passes, undefined).pipe(
+                  Effect.as({
+                    schemaVersion: 2,
+                    snapshotSequence: 1,
+                    threads: [idle, active],
+                    archivedThreads: [],
+                  }),
+                ),
+            }),
+            Layer.mock(ProjectStore.ProjectStoreV2)({
+              listShells: () => Effect.succeed([project(idle), project(active)]),
+            }),
+            Layer.mock(GitManager.GitManager)({}),
+            Layer.mock(PullRequestService.PullRequestService)({}),
+            // No repository identity: the group ends right after this lookup.
+            Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
+              resolve: (root) =>
+                Effect.sync(() => {
+                  lookups.set(root, (lookups.get(root) ?? 0) + 1);
+                  return null;
+                }),
+            }),
+            Layer.succeed(ServerActivation.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* () {
+            yield* TestClock.setTime(NOW_MS);
+            const service = yield* ThreadPullRequestService.make;
+            yield* service.start();
+            yield* Deferred.succeed(activation, undefined);
+            // The startup pass, then one pass a minute.
+            yield* Queue.take(passes);
+            yield* service.drain;
+            for (let pass = 1; pass < IDLE_SWEEP_EVERY_PASSES; pass += 1) {
+              yield* TestClock.adjust("1 minute");
+              yield* Queue.take(passes);
+              yield* service.drain;
+            }
+            expect(lookups.get("/workspace/active")).toBe(IDLE_SWEEP_EVERY_PASSES);
+            expect(lookups.get("/workspace/idle")).toBe(1);
+          }).pipe(Effect.provide(layerDependencies));
+        }),
+      ),
+  );
+});

apps/server/src/orchestration-v2/threadPullRequestSweep.expbkt3.ts

@@ -0,0 +1,57 @@
+/**
+ * T3-CUSTOM(expbkt3): which threads a periodic pull-request sweep checks.
+ *
+ * Each check starts several git processes, and every start forks the whole
+ * server process on the event loop. With 1,100 unsettled threads the sweep ran
+ * for 50-95 s of every two minutes and kept the loop saturated, so clients
+ * timed out while connecting. A thread with no run or message for a day is
+ * now checked on a rotation, once every IDLE_SWEEP_EVERY_PASSES all-thread
+ * passes. Its own events (a run ending, a checkpoint, an unsettle) still queue
+ * an immediate one-thread check, so activity is never waited on.
+ */
+import type { OrchestrationV2ThreadShell } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+
+export const IDLE_SWEEP_AFTER_MS = 24 * 60 * 60 * 1000;
+export const IDLE_SWEEP_EVERY_PASSES = 30;
+
+type SweepThread = Pick<
+  OrchestrationV2ThreadShell,
+  | "id"
+  | "activeRunId"
+  | "activityRunStatus"
+  | "createdAt"
+  | "latestUserMessageAt"
+  | "latestRunRequestedAt"
+  | "latestRunStartedAt"
+  | "latestRunCompletedAt"
+>;
+
+const latestActivityMs = (thread: SweepThread) =>
+  Math.max(
+    ...[
+      thread.createdAt,
+      thread.latestUserMessageAt,
+      thread.latestRunRequestedAt,
+      thread.latestRunStartedAt,
+      thread.latestRunCompletedAt,
+    ].map((at) => (at == null ? Number.NEGATIVE_INFINITY : DateTime.toEpochMillis(at))),
+  );
+
+export const isIdleForSweep = (thread: SweepThread, nowMs: number) =>
+  thread.activeRunId === null &&
+  thread.activityRunStatus == null &&
+  nowMs - latestActivityMs(thread) >= IDLE_SWEEP_AFTER_MS;
+
+/** The pass, modulo IDLE_SWEEP_EVERY_PASSES, on which an idle thread is checked. */
+export const idleSweepSlot = (threadId: string) => {
+  let hash = 0;
+  for (let index = 0; index < threadId.length; index += 1) {
+    hash = (Math.imul(hash, 31) + threadId.charCodeAt(index)) >>> 0;
+  }
+  return hash % IDLE_SWEEP_EVERY_PASSES;
+};
+
+/** Whether all-thread pass number `pass` checks this thread. */
+export const includeInPeriodicSweep = (thread: SweepThread, nowMs: number, pass: number) =>
+  !isIdleForSweep(thread, nowMs) || idleSweepSlot(thread.id) === pass % IDLE_SWEEP_EVERY_PASSES;