diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 2ded30eabfcd7..5b3a82f1ea04a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -2716,18 +2716,31 @@ export const layer: Layer.Layer = 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` 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` diff --git a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts b/apps/server/src/orchestration-v2/ThreadPullRequestService.ts index cd88d93977768..a1dabb267b2f8 100644 --- a/apps/server/src/orchestration-v2/ThreadPullRequestService.ts +++ b/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(); + // T3-CUSTOM(expbkt3): counts all-thread passes for the idle-thread rotation. + let periodicPass = 0; const finishBackfill = (threads: ReadonlyArray>) => { 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 || diff --git a/apps/server/src/orchestration-v2/threadPullRequestSweep.expbkt3.test.ts b/apps/server/src/orchestration-v2/threadPullRequestSweep.expbkt3.test.ts new file mode 100644 index 0000000000000..cd5fda6ed57d2 --- /dev/null +++ b/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 => { + 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(); + 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(); + const passes = yield* Queue.unbounded(); + const lookups = new Map(); + 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)); + }), + ), + ); +}); diff --git a/apps/server/src/orchestration-v2/threadPullRequestSweep.expbkt3.ts b/apps/server/src/orchestration-v2/threadPullRequestSweep.expbkt3.ts new file mode 100644 index 0000000000000..b2014a0f10c30 --- /dev/null +++ b/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;