authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-25 20:31:29-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-25 20:31:29-07:00
log3b0a495b0e05d11dbf88ff2148e7c21b49508172
tree8bdd9b829c51d7791b2840f8909b9d5afb24bb72
parent6989856aa630611dd7d6dcc8cfe9f97456816284
signature Signed by PGP key B5690EEEBB952194

perf(server): per-thread settlement and PR checks no longer rebuild the whole thread list (#13691)

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

5 files changed, 173 insertions(+), 21 deletions(-)

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+69
......@@ -30,6 +30,7 @@ import * as ThreadPlanProgress from "../ThreadPlanProgress.ts";
3030import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";
3131import { encodeThreadDetailPageCursor } from "../threadDetailCursor.ts";
3232import { projectThreadDetailSnapshot } from "../ActivityPayloadProjection.ts";
33import { readSweepSnapshot } from "../ThreadPullRequestReactor.ts";
3334import { makeSqlStatementCounter } from "../../../integration/SqlStatementCounter.integration.ts";
3435
3536const asProjectId = (value: string): ProjectId => ProjectId.make(value);
......@@ -3574,6 +3575,74 @@ it.effect(
35743575 },
35753576);
35763577
3578it.effect("reads one sweep thread and its projects like the shell snapshot", () => {
3579 const layer = OrchestrationProjectionSnapshotQueryLive.pipe(
3580 Layer.provide(ThreadBackgroundLiveness.layer),
3581 Layer.provide(ThreadPlanProgress.layer),
3582 Layer.provide(
3583 Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, {
3584 resolve: () =>
3585 Effect.succeed({
3586 canonicalKey: "github.com/acme/web",
3587 provider: "github",
3588 displayName: "acme/web",
3589 locator: {
3590 source: "git-remote" as const,
3591 remoteName: "origin",
3592 remoteUrl: "https://github.com/acme/web.git",
3593 },
3594 }),
3595 }),
3596 ),
3597 Layer.provideMerge(SqlitePersistenceMemory),
3598 );
3599 return Effect.gen(function* () {
3600 const sql = yield* SqlClient.SqlClient;
3601 const query = yield* ProjectionSnapshotQuery;
3602 yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at)
3603 VALUES ('p1', 'One', '/one', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z'),
3604 ('p2', 'Two', '/two', '[]', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z'),
3605 ('p3', 'Three', '/three', '[]', '2026-09-03T00:00:00Z', '2026-09-03T00:00:00Z')`;
3606 yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, branch, worktree_path, branch_pull_request_json, latest_turn_id, latest_user_message_at, pending_approval_count, snoozed_until, snoozed_at, created_at, updated_at, settled_override, settled_at)
3607 VALUES
3608 ('t-linked', 'p1', 'Linked', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'feature', '/one/wt', NULL, 'turn-1', '2026-09-02T00:00:00Z', 1, NULL, NULL, '2026-09-01T00:00:00Z', '2026-09-02T00:00:00Z', NULL, NULL),
3609 ('t-branch', 'p1', 'Branch', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'other', NULL,
3610 '{"projectId":"p2","repository":"acme/web","number":8,"url":"https://github.com/acme/web/pull/8"}',
3611 NULL, NULL, 0, '2026-09-10T00:00:00Z', '2026-09-02T00:00:00Z', '2026-09-01T00:00:00Z', '2026-09-02T00:00:00Z', 'settled', '2026-09-03T00:00:00Z'),
3612 ('t-other', 'p3', 'Other', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, NULL, NULL, NULL, 0, NULL, NULL, '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, NULL)`;
3613 yield* sql`INSERT INTO projection_thread_pull_requests (thread_id, host, repository, number, url, source, linked_at)
3614 VALUES ('t-linked', 'github.com', 'acme/web', 7, 'https://github.com/acme/web/pull/7', 'agent', '2026-09-02T00:00:00Z')`;
3615 yield* sql`INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, started_at, completed_at, checkpoint_files_json)
3616 VALUES ('t-linked', 'turn-1', 'completed', '2026-09-02T00:00:00Z', '2026-09-02T00:00:01Z', '2026-09-02T00:00:02Z', '[]')`;
3617 yield* sql`INSERT INTO projection_thread_sessions (thread_id, status, provider_name, active_turn_id, last_error, updated_at)
3618 VALUES ('t-linked', 'ready', 'codex', NULL, NULL, '2026-09-02T00:00:03Z')`;
3619 for (const projector of Object.values(ORCHESTRATION_PROJECTOR_NAMES)) {
3620 yield* sql`INSERT INTO projection_state (projector, last_applied_sequence, updated_at)
3621 VALUES (${projector}, 9, '2026-09-02T00:00:03Z')`;
3622 }
3623
3624 const full = yield* query.getShellSnapshot();
3625 // The seeded fields must reach the snapshot, or the parity check is empty.
3626 const linked = full.threads.find((thread) => thread.id === ThreadId.make("t-linked"));
3627 assert.strictEqual(full.snapshotSequence, 9);
3628 assert.strictEqual(linked?.linkedPullRequest?.number, 7);
3629 assert.strictEqual(linked?.latestTurn?.turnId, asTurnId("turn-1"));
3630 assert.strictEqual(linked?.session?.status, "ready");
3631
3632 for (const [threadId, projectIds] of [
3633 [ThreadId.make("t-linked"), [asProjectId("p1")]],
3634 // Settlement also needs the project that the saved branch PR names.
3635 [ThreadId.make("t-branch"), [asProjectId("p1"), asProjectId("p2")]],
3636 ] as const) {
3637 assert.deepStrictEqual(yield* readSweepSnapshot(query, threadId), {
3638 snapshotSequence: full.snapshotSequence,
3639 projects: full.projects.filter((project) => projectIds.includes(project.id)),
3640 threads: full.threads.filter((thread) => thread.id === threadId),
3641 });
3642 }
3643 }).pipe(Effect.provide(layer));
3644});
3645
35773646projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {
35783647 it.effect("lists one kind across active threads only, without hydrating the threads", () =>
35793648 Effect.gen(function* () {
apps/server/src/orchestration/ThreadPullRequestReactor.test.ts+39-4
......@@ -21,6 +21,7 @@ import * as Deferred from "effect/Deferred";
2121import * as Effect from "effect/Effect";
2222import * as FileSystem from "effect/FileSystem";
2323import * as Layer from "effect/Layer";
24import * as Option from "effect/Option";
2425import * as PubSub from "effect/PubSub";
2526import * as Queue from "effect/Queue";
2627import * as Ref from "effect/Ref";
......@@ -142,7 +143,8 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options
142143 threads: options.threads,
143144 updatedAt: NOW,
144145 });
145 const reads = yield* Queue.unbounded<void>();
146 // Each shell read: a thread id for a one-thread read, null for a full read.
147 const reads = yield* Queue.unbounded<ThreadId | null>();
146148 const events = yield* PubSub.unbounded<OrchestrationEvent>();
147149 const commands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);
148150 const branchCalls = yield* Ref.make<
......@@ -152,8 +154,24 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options
152154 let uuid = 0;
153155 const dependencies = Layer.mergeAll(
154156 Layer.mock(ProjectionSnapshotQuery)({
155 getShellSnapshot: () =>
156 Ref.get(snapshots).pipe(Effect.tap(() => Queue.offer(reads, undefined))),
157 getShellSnapshot: () => Ref.get(snapshots).pipe(Effect.tap(() => Queue.offer(reads, null))),
158 getSnapshotSequence: () =>
159 Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))),
160 getThreadShellById: (threadId) =>
161 Ref.get(snapshots).pipe(
162 Effect.map(({ threads }) =>
163 Option.fromUndefinedOr(
164 threads.find((thread) => thread.id === threadId && thread.archivedAt === null),
165 ),
166 ),
167 Effect.tap(() => Queue.offer(reads, threadId)),
168 ),
169 getProjectShells: (projectIds) =>
170 Ref.get(snapshots).pipe(
171 Effect.map(({ projects }) =>
172 projects.filter((project) => projectIds?.includes(project.id) ?? true),
173 ),
174 ),
157175 }),
158176 Layer.mock(GitManager)({
159177 branchPullRequest: (input, readOptions) =>
......@@ -376,7 +394,7 @@ describe("ThreadPullRequestReactor", () => {
376394 : [checkpointEvent, sessionEvent];
377395 for (const event of events) {
378396 yield* fixture.publish(event);
379 yield* Queue.take(fixture.reads);
397 expect(yield* Queue.take(fixture.reads)).toBe(current.id);
380398 yield* reactor.drain;
381399 }
382400 expect((yield* Ref.get(fixture.commands))[0]?.branchPullRequest).toEqual(reference(42));
......@@ -516,6 +534,23 @@ describe("ThreadPullRequestReactor", () => {
516534 yield* Effect.gen(function* () {
517535 const reactor = yield* fixture.start();
518536 expect(yield* Ref.get(fixture.commands)).toHaveLength(0);
537 // A one-thread read cannot show that other pending threads are gone.
538 const gone = ThreadId.make("gone");
539 yield* fixture.publish({
540 type: "thread.unarchived",
541 sequence: 2,
542 eventId: EventId.make("gone-unarchived"),
543 aggregateKind: "thread",
544 aggregateId: gone,
545 occurredAt: NOW,
546 commandId: null,
547 causationEventId: null,
548 correlationId: null,
549 metadata: {},
550 payload: { threadId: gone, updatedAt: NOW },
551 });
552 expect(yield* Queue.take(fixture.reads)).toBe(gone);
553 yield* reactor.drain;
519554 yield* Ref.set(online, true);
520555 yield* TestClock.adjust("1 minute");
521556 yield* Queue.take(fixture.reads);
apps/server/src/orchestration/ThreadPullRequestReactor.ts+36-3
......@@ -6,6 +6,7 @@ import {
66 CommandId,
77 type OrchestrationEvent,
88 type OrchestrationProjectShell,
9 type OrchestrationShellSnapshot,
910 type ThreadId,
1011 type ThreadLinkedPullRequest,
1112} from "@t3tools/contracts";
......@@ -16,11 +17,13 @@ import * as Crypto from "effect/Crypto";
1617import * as Effect from "effect/Effect";
1718import * as FileSystem from "effect/FileSystem";
1819import * as Layer from "effect/Layer";
20import * as Option from "effect/Option";
1921import * as Schedule from "effect/Schedule";
2022import type * as Scope from "effect/Scope";
2123import * as Stream from "effect/Stream";
2224
2325import * as GitManager from "../git/GitManager.ts";
26import type { ProjectionRepositoryError } from "../persistence/Errors.ts";
2427import * as PullRequestService from "../pullRequest/PullRequestService.ts";
2528import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
2629import { forkParked } from "../serverActivation.ts";
......@@ -69,6 +72,35 @@ export function pullRequestMatchesProject(
6972 );
7073}
7174
75/**
76 * Read the shell state for a discovery or settlement sweep. A sweep for one
77 * thread reads that thread and the projects it names, not every thread.
78 */
79export const readSweepSnapshot = (
80 snapshots: ProjectionSnapshotQuery.ProjectionSnapshotQueryShape,
81 threadId: ThreadId | null,
82): Effect.Effect<
83 Pick<OrchestrationShellSnapshot, "snapshotSequence" | "projects" | "threads">,
84 ProjectionRepositoryError
85> =>
86 threadId === null
87 ? snapshots.getShellSnapshot()
88 : Effect.gen(function* () {
89 // Read the sequence first. The thread is then at least this new, so a
90 // command guarded by the sequence is rejected rather than missing a change.
91 const { snapshotSequence } = yield* snapshots.getSnapshotSequence();
92 const thread = yield* snapshots.getThreadShellById(threadId);
93 if (Option.isNone(thread)) return { snapshotSequence, projects: [], threads: [] };
94 // Settlement also checks the project a saved pull request names.
95 const reference = thread.value.linkedPullRequest ?? thread.value.branchPullRequest;
96 const projects = yield* snapshots.getProjectShells(
97 reference == null
98 ? [thread.value.projectId]
99 : [thread.value.projectId, reference.projectId],
100 );
101 return { snapshotSequence, projects, threads: [thread.value] };
102 });
103
72104/** @public Service construction is part of the canonical Effect module API. */
73105export const make = Effect.gen(function* () {
74106 const engine = yield* OrchestrationEngine.OrchestrationEngineService;
......@@ -97,7 +129,7 @@ export const make = Effect.gen(function* () {
97129 const synchronize = Effect.fn("ThreadPullRequestReactor.synchronize")(function* (
98130 request: RefreshRequest,
99131 ) {
100 const snapshot = yield* snapshots.getShellSnapshot();
132 const snapshot = yield* readSweepSnapshot(snapshots, request.threadId);
101133 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
102134 if (request.backfill) {
103135 for (const thread of snapshot.threads) {
......@@ -109,14 +141,15 @@ export const make = Effect.gen(function* () {
109141 }
110142 }
111143 }
144 // A single-thread read only shows whether its own thread is gone.
112145 const threadIds = new Set(snapshot.threads.map((thread) => thread.id));
113 for (const threadId of pendingBackfill.keys()) {
146 const checkedIds = request.threadId === null ? pendingBackfill.keys() : [request.threadId];
147 for (const threadId of checkedIds) {
114148 if (!threadIds.has(threadId)) pendingBackfill.delete(threadId);
115149 }
116150 const threads = snapshot.threads.filter(
117151 (thread) =>
118152 thread.archivedAt === null &&
119 (request.threadId === null || thread.id === request.threadId) &&
120153 ((thread.settledOverride !== "settled" && thread.settledAt === null) ||
121154 request.threadId !== null ||
122155 pendingBackfill.has(thread.id)) &&
apps/server/src/orchestration/ThreadSettlementReactor.test.ts+25-6
......@@ -24,6 +24,7 @@ import * as Effect from "effect/Effect";
2424import * as Fiber from "effect/Fiber";
2525import * as FileSystem from "effect/FileSystem";
2626import * as Layer from "effect/Layer";
27import * as Option from "effect/Option";
2728import * as PubSub from "effect/PubSub";
2829import * as Queue from "effect/Queue";
2930import * as Ref from "effect/Ref";
......@@ -181,7 +182,8 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
181182 const activation = yield* Deferred.make<void>();
182183 const snapshots = yield* Ref.make(options.snapshot);
183184 const snapshotReadCount = yield* Ref.make(0);
184 const snapshotReads = yield* Queue.unbounded<number>();
185 // Each shell read: a thread id for a one-thread read, null for a full read.
186 const snapshotReads = yield* Queue.unbounded<ThreadId | null>();
185187 const settings = yield* Ref.make(options.settings ?? DEFAULT_SERVER_SETTINGS);
186188 const settingsReads = yield* Queue.unbounded<ServerSettings>();
187189 const settingsChanges = yield* PubSub.unbounded<ServerSettings>();
......@@ -255,10 +257,27 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
255257 const dependencies = Layer.mergeAll(
256258 Layer.mock(ProjectionSnapshotQuery)({
257259 getShellSnapshot: () =>
258 Ref.updateAndGet(snapshotReadCount, (count) => count + 1).pipe(
259 Effect.tap((count) => Queue.offer(snapshotReads, count)),
260 Ref.update(snapshotReadCount, (count) => count + 1).pipe(
261 Effect.andThen(Queue.offer(snapshotReads, null)),
260262 Effect.andThen(Ref.get(snapshots)),
261263 ),
264 getSnapshotSequence: () =>
265 Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))),
266 getThreadShellById: (threadId) =>
267 Ref.get(snapshots).pipe(
268 Effect.map(({ threads }) =>
269 Option.fromUndefinedOr(
270 threads.find((thread) => thread.id === threadId && thread.archivedAt === null),
271 ),
272 ),
273 Effect.tap(() => Queue.offer(snapshotReads, threadId)),
274 ),
275 getProjectShells: (projectIds) =>
276 Ref.get(snapshots).pipe(
277 Effect.map(({ projects }) =>
278 projects.filter((project) => projectIds?.includes(project.id) ?? true),
279 ),
280 ),
262281 }),
263282 Layer.mock(GitManager)({
264283 branchPullRequest,
......@@ -313,7 +332,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
313332const startHarness = Effect.fn("startThreadSettlementHarness")(function* (
314333 reactor: ThreadSettlementReactor.ThreadSettlementReactor["Service"],
315334 activation: Deferred.Deferred<void>,
316 snapshotReads: Queue.Queue<number>,
335 snapshotReads: Queue.Queue<ThreadId | null>,
317336) {
318337 yield* reactor.start();
319338 yield* Deferred.succeed(activation, undefined);
......@@ -443,7 +462,7 @@ describe("ThreadSettlementReactor", () => {
443462 updatedAt: NOW,
444463 },
445464 });
446 yield* Queue.take(fixture.snapshotReads);
465 assert.strictEqual(yield* Queue.take(fixture.snapshotReads), thread.id);
447466 yield* reactor.drain;
448467 }
449468 assert.deepStrictEqual(
......@@ -463,7 +482,7 @@ describe("ThreadSettlementReactor", () => {
463482 aggregateId: readySession.threadId,
464483 payload: { threadId: readySession.threadId, session: readySession },
465484 });
466 yield* Queue.take(fixture.snapshotReads);
485 assert.strictEqual(yield* Queue.take(fixture.snapshotReads), readySession.threadId);
467486 yield* reactor.drain;
468487 assert.deepStrictEqual(
469488 (yield* Ref.get(fixture.commands)).map(({ threadId }) => threadId),
apps/server/src/orchestration/ThreadSettlementReactor.ts+4-8
......@@ -23,7 +23,7 @@ import * as ServerSettings from "../serverSettings.ts";
2323import { forkParked } from "../serverActivation.ts";
2424import * as OrchestrationEngine from "./Services/OrchestrationEngine.ts";
2525import * as ProjectionSnapshotQuery from "./Services/ProjectionSnapshotQuery.ts";
26import { pullRequestMatchesProject } from "./ThreadPullRequestReactor.ts";
26import { pullRequestMatchesProject, readSweepSnapshot } from "./ThreadPullRequestReactor.ts";
2727import {
2828 isAutoSettlementCandidate,
2929 resolveAutoSettlementAt,
......@@ -95,20 +95,16 @@ export const make = Effect.gen(function* () {
9595 if (!autoSettlementConfigured(settings)) {
9696 return;
9797 }
98 const snapshot = yield* snapshots.getShellSnapshot();
98 const snapshot = yield* readSweepSnapshot(snapshots, threadId ?? null);
9999 const now = DateTime.formatIso(yield* DateTime.now);
100100 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
101101 // A merge rechecks all candidates, including branches that discovery has
102102 // not linked yet. Those lookups can still have cached the PR as open.
103 const candidates = snapshot.threads.filter(
104 (thread) =>
105 (threadId === undefined || thread.id === threadId) &&
106 isAutoSettlementCandidate(thread, now),
107 );
103 const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));
108104
109105 // Return the thread when it still needs a pull request decision. A rejected
110106 // dispatch skips it for this snapshot instead of retrying through a lookup.
111 const settleThread = Effect.fn("ThreadSettlementReactor.settleThread")(
107 const settleThread = Effect.fnUntraced(
112108 function* (thread: (typeof candidates)[number], pullRequest: SettlementPullRequest | null) {
113109 const settings = resolveProjectSettings(
114110 yield* settingsService.getSettings,