authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-25 19:29:13-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-25 19:29:13-07:00
log6530de0339d2ca49957d0039133c49e3a08557f7
tree5e98cd0e6938f1c38e09c4df3a23b464684e2403
parent75bf92e652f72cb0062c102236aeb130d1d714b3
signature Signed by PGP key B5690EEEBB952194

perf(server): pull request sync reads only threads with linked pull requests (#13704)

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

12 files changed, 187 insertions(+), 10 deletions(-)

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts+5
...@@ -84,6 +84,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -84,6 +84,7 @@ describe("CheckpointDiffQuery.layer", () => {
84 getShellSnapshot: () =>84 getShellSnapshot: () =>
85 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),85 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
86 getDeletedWorktreeThreads: () => Effect.die("unused"),86 getDeletedWorktreeThreads: () => Effect.die("unused"),
87 listThreadsWithPullRequests: () => Effect.die("unused"),
87 getArchivedShellSnapshot: () =>88 getArchivedShellSnapshot: () =>
88 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),89 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
89 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),90 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
...@@ -201,6 +202,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -201,6 +202,7 @@ describe("CheckpointDiffQuery.layer", () => {
201 getShellSnapshot: () =>202 getShellSnapshot: () =>
202 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),203 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
203 getDeletedWorktreeThreads: () => Effect.die("unused"),204 getDeletedWorktreeThreads: () => Effect.die("unused"),
205 listThreadsWithPullRequests: () => Effect.die("unused"),
204 getArchivedShellSnapshot: () =>206 getArchivedShellSnapshot: () =>
205 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),207 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
206 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),208 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
...@@ -293,6 +295,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -293,6 +295,7 @@ describe("CheckpointDiffQuery.layer", () => {
293 getShellSnapshot: () =>295 getShellSnapshot: () =>
294 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),296 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
295 getDeletedWorktreeThreads: () => Effect.die("unused"),297 getDeletedWorktreeThreads: () => Effect.die("unused"),
298 listThreadsWithPullRequests: () => Effect.die("unused"),
296 getArchivedShellSnapshot: () =>299 getArchivedShellSnapshot: () =>
297 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),300 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
298 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),301 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
...@@ -370,6 +373,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -370,6 +373,7 @@ describe("CheckpointDiffQuery.layer", () => {
370 getShellSnapshot: () =>373 getShellSnapshot: () =>
371 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),374 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
372 getDeletedWorktreeThreads: () => Effect.die("unused"),375 getDeletedWorktreeThreads: () => Effect.die("unused"),
376 listThreadsWithPullRequests: () => Effect.die("unused"),
373 getArchivedShellSnapshot: () =>377 getArchivedShellSnapshot: () =>
374 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),378 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
375 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),379 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
...@@ -432,6 +436,7 @@ describe("CheckpointDiffQuery.layer", () => {...@@ -432,6 +436,7 @@ describe("CheckpointDiffQuery.layer", () => {
432 getShellSnapshot: () =>436 getShellSnapshot: () =>
433 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),437 Effect.die("CheckpointDiffQuery should not request the orchestration shell snapshot"),
434 getDeletedWorktreeThreads: () => Effect.die("unused"),438 getDeletedWorktreeThreads: () => Effect.die("unused"),
439 listThreadsWithPullRequests: () => Effect.die("unused"),
435 getArchivedShellSnapshot: () =>440 getArchivedShellSnapshot: () =>
436 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),441 Effect.die("CheckpointDiffQuery should not request archived shell snapshots"),
437 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),442 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }),
apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+1
...@@ -435,6 +435,7 @@ describe("OrchestrationEngine", () => {...@@ -435,6 +435,7 @@ describe("OrchestrationEngine", () => {
435 updatedAt: projectionSnapshot.updatedAt,435 updatedAt: projectionSnapshot.updatedAt,
436 }),436 }),
437 getDeletedWorktreeThreads: () => Effect.die("unused"),437 getDeletedWorktreeThreads: () => Effect.die("unused"),
438 listThreadsWithPullRequests: () => Effect.die("unused"),
438 getArchivedShellSnapshot: () =>439 getArchivedShellSnapshot: () =>
439 Effect.succeed({440 Effect.succeed({
440 snapshotSequence: projectionSnapshot.snapshotSequence,441 snapshotSequence: projectionSnapshot.snapshotSequence,
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+71
...@@ -3503,6 +3503,77 @@ it.effect("omits foreign-host PRs from legacy snapshots while preserving native...@@ -3503,6 +3503,77 @@ it.effect("omits foreign-host PRs from legacy snapshots while preserving native
3503 }).pipe(Effect.provide(layer));3503 }).pipe(Effect.provide(layer));
3504});3504});
35053505
3506it.effect(
3507 "lists linked threads like the shell snapshot, in one query and without identities",
3508 () => {
3509 const resolved: string[] = [];
3510 const layer = OrchestrationProjectionSnapshotQueryLive.pipe(
3511 Layer.provide(ThreadBackgroundLiveness.layer),
3512 Layer.provide(ThreadPlanProgress.layer),
3513 Layer.provide(
3514 Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, {
3515 resolve: (root) =>
3516 Effect.sync(() => {
3517 resolved.push(root);
3518 return null;
3519 }),
3520 }),
3521 ),
3522 Layer.provideMerge(SqlitePersistenceMemory),
3523 );
3524 return Effect.gen(function* () {
3525 const sql = yield* SqlClient.SqlClient;
3526 const query = yield* ProjectionSnapshotQuery;
3527 yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at)
3528 VALUES ('p1', 'One', '/one', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z'),
3529 ('p2', 'Two', '/two', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z')`;
3530 yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, created_at, updated_at, archived_at, deleted_at, settled_override, settled_at)
3531 VALUES
3532 ('t-late', 'p1', 'Late', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-03T00:00:00Z', '2026-09-03T00:00:00Z', NULL, NULL, 'settled', '2026-09-04T00:00:00Z'),
3533 ('t-early', 'p2', 'Early', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, NULL, NULL, NULL),
3534 ('t-first', 'p1', 'First', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', NULL, NULL, NULL, NULL),
3535 ('t-plain', 'p1', 'Plain', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', NULL, NULL, NULL, NULL),
3536 ('t-archived', 'p1', 'Archived', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', '2026-09-05T00:00:00Z', NULL, NULL, NULL),
3537 ('t-deleted', 'p1', 'Deleted', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z', NULL, '2026-09-05T00:00:00Z', NULL, NULL)`;
3538 yield* sql`INSERT INTO projection_thread_pull_requests (thread_id, host, repository, number, url, source, linked_at, snapshot_json)
3539 VALUES
3540 ('t-late', 'github.com', 'acme/web', 3, 'https://github.com/acme/web/pull/3', 'manual', '2026-09-03T00:00:00Z', NULL),
3541 ('t-early', 'github.com', 'acme/api', 4, 'https://github.com/acme/api/pull/4', 'agent', '2026-09-01T00:00:00Z', NULL),
3542 ('t-first', 'github.com', 'acme/web', 2, 'https://github.com/acme/web/pull/2', 'stack-dismissed', '2026-09-02T00:00:00Z', NULL),
3543 ('t-first', 'github.com', 'acme/web', 1, 'https://github.com/acme/web/pull/1', 'created', '2026-09-02T00:00:00Z',
3544 '{"state":"open","title":"One","headBranch":"one","baseBranch":"main","isDraft":false,"updatedAt":null,"syncedAt":"2026-09-02T00:00:00Z"}'),
3545 ('t-archived', 'github.com', 'acme/web', 5, 'https://github.com/acme/web/pull/5', 'manual', '2026-09-02T00:00:00Z', NULL),
3546 ('t-deleted', 'github.com', 'acme/web', 6, 'https://github.com/acme/web/pull/6', 'manual', '2026-09-02T00:00:00Z', NULL)`;
3547 const expected = (yield* query.getShellSnapshot()).threads
3548 .filter((thread) => thread.pullRequests.length > 0)
3549 .map(({ id, projectId, settledOverride, settledAt, pullRequests }) => ({
3550 id,
3551 projectId,
3552 settledOverride,
3553 settledAt,
3554 pullRequests,
3555 }));
3556 resolved.length = 0;
3557
3558 const counter = makeSqlStatementCounter();
3559 const threads = yield* query
3560 .listThreadsWithPullRequests()
3561 .pipe(Effect.withTracer(counter.tracer));
3562 assert.deepStrictEqual(
3563 threads.map((thread) => [thread.id, thread.pullRequests.map((link) => link.number)]),
3564 [
3565 ["t-first", [1, 2]],
3566 ["t-late", [3]],
3567 ["t-early", [4]],
3568 ],
3569 );
3570 assert.deepStrictEqual(threads, expected);
3571 assert.strictEqual(counter.count(), 1);
3572 assert.deepStrictEqual(resolved, []);
3573 }).pipe(Effect.provide(layer));
3574 },
3575);
3576
3506projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {3577projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {
3507 it.effect("lists one kind across active threads only, without hydrating the threads", () =>3578 it.effect("lists one kind across active threads only, without hydrating the threads", () =>
3508 Effect.gen(function* () {3579 Effect.gen(function* () {
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+66
...@@ -77,6 +77,7 @@ import {...@@ -77,6 +77,7 @@ import {
77 type ProjectionSnapshotCounts,77 type ProjectionSnapshotCounts,
78 type ProjectionThreadCheckpointContext,78 type ProjectionThreadCheckpointContext,
79 type ProjectionThreadDetailQuery,79 type ProjectionThreadDetailQuery,
80 type ProjectionThreadPullRequests,
80 type ProjectionSnapshotQueryShape,81 type ProjectionSnapshotQueryShape,
81} from "../Services/ProjectionSnapshotQuery.ts";82} from "../Services/ProjectionSnapshotQuery.ts";
8283
...@@ -802,6 +803,41 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -802,6 +803,41 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
802 `,803 `,
803 });804 });
804805
806 // One row per link, in the shell snapshot's thread order and link order.
807 const listActiveThreadPullRequestSyncRows = SqlSchema.findAll({
808 Request: Schema.Void,
809 Result: ProjectionThreadPullRequestDbRowSchema.mapFields(
810 Struct.assign({
811 projectId: ProjectionThread.fields.projectId,
812 settledOverride: ProjectionThread.fields.settledOverride,
813 settledAt: ProjectionThread.fields.settledAt,
814 }),
815 ),
816 execute: () =>
817 sql`
818 SELECT
819 links.thread_id AS "threadId",
820 threads.project_id AS "projectId",
821 threads.settled_override AS "settledOverride",
822 threads.settled_at AS "settledAt",
823 links.host,
824 links.repository,
825 links.number,
826 links.url,
827 links.source,
828 links.linked_at AS "linkedAt",
829 links.snapshot_json AS "snapshot",
830 links.stack_json AS "stack"
831 FROM projection_thread_pull_requests links
832 INNER JOIN projection_threads threads
833 ON threads.thread_id = links.thread_id
834 WHERE threads.deleted_at IS NULL
835 AND threads.archived_at IS NULL
836 ORDER BY threads.project_id ASC, threads.created_at ASC, threads.thread_id ASC,
837 links.linked_at ASC, links.number ASC
838 `,
839 });
840
805 const listArchivedThreadPullRequestRows = SqlSchema.findAll({841 const listArchivedThreadPullRequestRows = SqlSchema.findAll({
806 Request: Schema.Void,842 Request: Schema.Void,
807 Result: ProjectionThreadPullRequestDbRowSchema,843 Result: ProjectionThreadPullRequestDbRowSchema,
...@@ -2783,6 +2819,35 @@ pending_approval_requests AS (...@@ -2783,6 +2819,35 @@ pending_approval_requests AS (
2783 }),2819 }),
2784 );2820 );
27852821
2822 const listThreadsWithPullRequests: ProjectionSnapshotQueryShape["listThreadsWithPullRequests"] =
2823 () =>
2824 listActiveThreadPullRequestSyncRows(undefined).pipe(
2825 Effect.map((rows) => {
2826 const threads = new Map<
2827 ThreadId,
2828 ProjectionThreadPullRequests & { readonly pullRequests: Array<ThreadPullRequestLink> }
2829 >();
2830 for (const row of rows) {
2831 const thread = threads.get(row.threadId) ?? {
2832 id: row.threadId,
2833 projectId: row.projectId,
2834 settledOverride: row.settledOverride,
2835 settledAt: row.settledAt,
2836 pullRequests: [],
2837 };
2838 thread.pullRequests.push(mapPullRequestRow(row));
2839 threads.set(row.threadId, thread);
2840 }
2841 return [...threads.values()];
2842 }),
2843 Effect.mapError(
2844 toPersistenceSqlOrDecodeError(
2845 "ProjectionSnapshotQuery.listThreadsWithPullRequests:query",
2846 "ProjectionSnapshotQuery.listThreadsWithPullRequests:decodeRows",
2847 ),
2848 ),
2849 );
2850
2786 const getArchivedShellSnapshot: ProjectionSnapshotQueryShape["getArchivedShellSnapshot"] = () =>2851 const getArchivedShellSnapshot: ProjectionSnapshotQueryShape["getArchivedShellSnapshot"] = () =>
2787 sql2852 sql
2788 .withTransaction(2853 .withTransaction(
...@@ -3778,6 +3843,7 @@ pending_approval_requests AS (...@@ -3778,6 +3843,7 @@ pending_approval_requests AS (
3778 listActivitiesByKind,3843 listActivitiesByKind,
3779 getSnapshot,3844 getSnapshot,
3780 getShellSnapshot,3845 getShellSnapshot,
3846 listThreadsWithPullRequests,
3781 getArchivedShellSnapshot,3847 getArchivedShellSnapshot,
3782 getDeletedWorktreeThreads,3848 getDeletedWorktreeThreads,
3783 searchThreads,3849 searchThreads,
apps/server/src/orchestration/PullRequestSyncReactor.test.ts+13-1
...@@ -172,6 +172,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:...@@ -172,6 +172,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
172 const snapshots = yield* Ref.make(options.snapshot);172 const snapshots = yield* Ref.make(options.snapshot);
173 const events = yield* PubSub.unbounded<OrchestrationEvent>();173 const events = yield* PubSub.unbounded<OrchestrationEvent>();
174 const snapshotReads = yield* Queue.unbounded<void>();174 const snapshotReads = yield* Queue.unbounded<void>();
175 const shellSnapshotReads = yield* Ref.make(0);
175 const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);176 const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);
176 const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);177 const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);
177 const summaryCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);178 const summaryCalls = yield* Ref.make<ReadonlyArray<PullRequestRef>>([]);
...@@ -209,8 +210,16 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:...@@ -209,8 +210,16 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
209210
210 const dependencies = Layer.mergeAll(211 const dependencies = Layer.mergeAll(
211 Layer.mock(ProjectionSnapshotQuery)({212 Layer.mock(ProjectionSnapshotQuery)({
213 listThreadsWithPullRequests: () =>
214 Queue.offer(snapshotReads, undefined).pipe(
215 Effect.andThen(Ref.get(snapshots)),
216 Effect.map((snapshot) => snapshot.threads),
217 ),
212 getShellSnapshot: () =>218 getShellSnapshot: () =>
213 Queue.offer(snapshotReads, undefined).pipe(Effect.andThen(Ref.get(snapshots))),219 Ref.update(shellSnapshotReads, (count) => count + 1).pipe(
220 Effect.andThen(Queue.offer(snapshotReads, undefined)),
221 Effect.andThen(Ref.get(snapshots)),
222 ),
214 }),223 }),
215 Layer.mock(PullRequestService)({224 Layer.mock(PullRequestService)({
216 summary,225 summary,
...@@ -235,6 +244,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:...@@ -235,6 +244,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
235 activation,244 activation,
236 snapshots,245 snapshots,
237 snapshotReads,246 snapshotReads,
247 shellSnapshotReads,
238 syncCommands,248 syncCommands,
239 linkCommands,249 linkCommands,
240 summaryCalls,250 summaryCalls,
...@@ -511,6 +521,8 @@ describe("PullRequestSyncReactor", () => {...@@ -511,6 +521,8 @@ describe("PullRequestSyncReactor", () => {
511 ],521 ],
512 );522 );
513 assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1);523 assert.strictEqual((yield* Ref.get(fixture.stackCalls)).length, 1);
524 // Reads only linked threads, never the full shell snapshot of every thread.
525 assert.strictEqual(yield* Ref.get(fixture.shellSnapshotReads), 0);
514 }).pipe(Effect.provide(fixture.layer));526 }).pipe(Effect.provide(fixture.layer));
515 }),527 }),
516 ),528 ),
apps/server/src/orchestration/PullRequestSyncReactor.ts+7-9
...@@ -1,7 +1,6 @@...@@ -1,7 +1,6 @@
1import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl";1import { siblingPullRequestUrl } from "@t3tools/shared/changeRequestUrl";
2import {2import {
3 CommandId,3 CommandId,
4 type OrchestrationThreadShell,
5 type PullRequestSummary,4 type PullRequestSummary,
6 type ThreadPullRequestKey,5 type ThreadPullRequestKey,
7 type ThreadPullRequestLink,6 type ThreadPullRequestLink,
...@@ -36,7 +35,7 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;...@@ -36,7 +35,7 @@ const SLOW_SYNC_INTERVAL_MS = 15 * 60 * 1_000;
36type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;35type SnapshotFields = Omit<ThreadPullRequestSnapshot, "syncedAt">;
3736
38interface LinkEntry {37interface LinkEntry {
39 readonly thread: OrchestrationThreadShell;38 readonly thread: ProjectionSnapshotQuery.ProjectionThreadPullRequests;
40 readonly link: ThreadPullRequestLink;39 readonly link: ThreadPullRequestLink;
41}40}
4241
...@@ -104,15 +103,15 @@ function stacksEqual(...@@ -104,15 +103,15 @@ function stacksEqual(
104 );103 );
105}104}
106105
107function isUnsettled(thread: OrchestrationThreadShell): boolean {106function isUnsettled(thread: ProjectionSnapshotQuery.ProjectionThreadPullRequests): boolean {
108 return thread.settledOverride !== "settled" && thread.settledAt === null;107 return thread.settledOverride !== "settled" && thread.settledAt === null;
109}108}
110109
111/**110/**
112 * Keeps every thread ↔ pull request link's host snapshot current. One sweep a minute reads111 * Keeps every thread ↔ pull request link's host snapshot current. One sweep a minute reads
113 * the shell snapshot, groups visible links by pull request so the host is asked once per PR112 * only the active threads that have links, groups visible links by pull request so the host
114 * no matter how many threads share it, and writes back only what changed. Native stacks the113 * is asked once per PR no matter how many threads share it, and writes back only what
115 * host reports are auto-linked to the thread as `source: "stack"`.114 * changed. Native stacks the host reports are auto-linked to the thread as `source: "stack"`.
116 */115 */
117export class PullRequestSyncReactor extends Context.Service<116export class PullRequestSyncReactor extends Context.Service<
118 PullRequestSyncReactor,117 PullRequestSyncReactor,
...@@ -153,14 +152,13 @@ export const make = Effect.gen(function* () {...@@ -153,14 +152,13 @@ export const make = Effect.gen(function* () {
153 Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);152 Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);
154153
155 const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (requestedKey?: string) {154 const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (requestedKey?: string) {
156 const snapshot = yield* snapshots.getShellSnapshot();155 const threads = yield* snapshots.listThreadsWithPullRequests();
157 const now = yield* DateTime.now;156 const now = yield* DateTime.now;
158 const nowMs = DateTime.toEpochMillis(now);157 const nowMs = DateTime.toEpochMillis(now);
159 const nowIso = DateTime.formatIso(now);158 const nowIso = DateTime.formatIso(now);
160159
161 const groups = new Map<string, Array<LinkEntry>>();160 const groups = new Map<string, Array<LinkEntry>>();
162 for (const thread of snapshot.threads) {161 for (const thread of threads) {
163 if (thread.archivedAt !== null) continue;
164 for (const link of visibleThreadPullRequests(thread.pullRequests)) {162 for (const link of visibleThreadPullRequests(thread.pullRequests)) {
165 const key = threadPullRequestKeyOf(link);163 const key = threadPullRequestKeyOf(link);
166 const entries = groups.get(key) ?? [];164 const entries = groups.get(key) ?? [];
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+16
...@@ -64,6 +64,12 @@ export interface ProjectionFullThreadDiffContext {...@@ -64,6 +64,12 @@ export interface ProjectionFullThreadDiffContext {
64 readonly toCheckpointRef: CheckpointRef | null;64 readonly toCheckpointRef: CheckpointRef | null;
65}65}
6666
67/** The thread fields pull request sync reads, for a thread with at least one link. */
68export type ProjectionThreadPullRequests = Pick<
69 OrchestrationThreadShell,
70 "id" | "projectId" | "settledOverride" | "settledAt" | "pullRequests"
71>;
72
67export interface ProjectionThreadDetailQuery {73export interface ProjectionThreadDetailQuery {
68 /**74 /**
69 * Limit activities before SQLite returns and decodes their payloads.75 * Limit activities before SQLite returns and decodes their payloads.
...@@ -131,6 +137,16 @@ export interface ProjectionSnapshotQueryShape {...@@ -131,6 +137,16 @@ export interface ProjectionSnapshotQueryShape {
131 ProjectionRepositoryError137 ProjectionRepositoryError
132 >;138 >;
133139
140 /**
141 * Read active (not deleted, not archived) threads that have at least one pull
142 * request link, in shell snapshot order. Skips repository identity, so no
143 * legacy `linkedPullRequest` is derived.
144 */
145 readonly listThreadsWithPullRequests: () => Effect.Effect<
146 ReadonlyArray<ProjectionThreadPullRequests>,
147 ProjectionRepositoryError
148 >;
149
134 /** Durable worktree ownership retained after thread deletion, including across restarts. */150 /** Durable worktree ownership retained after thread deletion, including across restarts. */
135 readonly getDeletedWorktreeThreads: () => Effect.Effect<151 readonly getDeletedWorktreeThreads: () => Effect.Effect<
136 ReadonlyArray<{152 ReadonlyArray<{
apps/server/src/project/AgentSessionScanner.test.ts+1
...@@ -48,6 +48,7 @@ const makeProjectionSnapshotQueryLayer = (importedWorkspaceRoots: ReadonlyArray<...@@ -48,6 +48,7 @@ const makeProjectionSnapshotQueryLayer = (importedWorkspaceRoots: ReadonlyArray<
48 updatedAt: "2026-01-01T00:00:00.000Z",48 updatedAt: "2026-01-01T00:00:00.000Z",
49 }),49 }),
50 getDeletedWorktreeThreads: () => Effect.die("unused"),50 getDeletedWorktreeThreads: () => Effect.die("unused"),
51 listThreadsWithPullRequests: () => Effect.die("unused"),
51 getArchivedShellSnapshot: () => Effect.die("unused"),52 getArchivedShellSnapshot: () => Effect.die("unused"),
52 getSnapshotSequence: () => Effect.die("unused"),53 getSnapshotSequence: () => Effect.die("unused"),
53 getCounts: () => Effect.die("unused"),54 getCounts: () => Effect.die("unused"),
apps/server/src/project/ProjectSetupScriptRunner.test.ts+1
...@@ -34,6 +34,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>...@@ -34,6 +34,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
34 getSnapshot: () => Effect.die("unused"),34 getSnapshot: () => Effect.die("unused"),
35 getShellSnapshot: () => Effect.die("unused"),35 getShellSnapshot: () => Effect.die("unused"),
36 getDeletedWorktreeThreads: () => Effect.die("unused"),36 getDeletedWorktreeThreads: () => Effect.die("unused"),
37 listThreadsWithPullRequests: () => Effect.die("unused"),
37 getArchivedShellSnapshot: () => Effect.die("unused"),38 getArchivedShellSnapshot: () => Effect.die("unused"),
38 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }),39 getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }),
39 getCounts: () => Effect.die("unused"),40 getCounts: () => Effect.die("unused"),
apps/server/src/provider/Layers/ProviderService.test.ts+1
...@@ -4969,6 +4969,7 @@ describe("agent browser access", () => {...@@ -4969,6 +4969,7 @@ describe("agent browser access", () => {
4969 getSnapshot: () => Effect.die("unused"),4969 getSnapshot: () => Effect.die("unused"),
4970 getShellSnapshot: () => Effect.die("unused"),4970 getShellSnapshot: () => Effect.die("unused"),
4971 getDeletedWorktreeThreads: () => Effect.die("unused"),4971 getDeletedWorktreeThreads: () => Effect.die("unused"),
4972 listThreadsWithPullRequests: () => Effect.die("unused"),
4972 getArchivedShellSnapshot: () => Effect.die("unused"),4973 getArchivedShellSnapshot: () => Effect.die("unused"),
4973 getSnapshotSequence: () => Effect.die("unused"),4974 getSnapshotSequence: () => Effect.die("unused"),
4974 getCounts: () => Effect.die("unused"),4975 getCounts: () => Effect.die("unused"),
apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+1
...@@ -239,6 +239,7 @@ describe("ProviderSessionReaper", () => {...@@ -239,6 +239,7 @@ describe("ProviderSessionReaper", () => {
239 getSnapshot: () => Effect.die("unused"),239 getSnapshot: () => Effect.die("unused"),
240 getShellSnapshot: () => Effect.die("unused"),240 getShellSnapshot: () => Effect.die("unused"),
241 getDeletedWorktreeThreads: () => Effect.die("unused"),241 getDeletedWorktreeThreads: () => Effect.die("unused"),
242 listThreadsWithPullRequests: () => Effect.die("unused"),
242 getArchivedShellSnapshot: () => Effect.die("unused"),243 getArchivedShellSnapshot: () => Effect.die("unused"),
243 getSnapshotSequence: () =>244 getSnapshotSequence: () =>
244 Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }),245 Effect.succeed({ snapshotSequence: input.readModel.snapshotSequence }),
apps/server/src/serverRuntimeStartup.test.ts+4
...@@ -169,6 +169,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa...@@ -169,6 +169,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
169 getSnapshot: () => Effect.die("unused"),169 getSnapshot: () => Effect.die("unused"),
170 getShellSnapshot: () => Effect.die("unused"),170 getShellSnapshot: () => Effect.die("unused"),
171 getDeletedWorktreeThreads: () => Effect.die("unused"),171 getDeletedWorktreeThreads: () => Effect.die("unused"),
172 listThreadsWithPullRequests: () => Effect.die("unused"),
172 getArchivedShellSnapshot: () => Effect.die("unused"),173 getArchivedShellSnapshot: () => Effect.die("unused"),
173 getSnapshotSequence: () => Effect.die("unused"),174 getSnapshotSequence: () => Effect.die("unused"),
174 getCounts: () => Effect.die("unused"),175 getCounts: () => Effect.die("unused"),
...@@ -298,6 +299,7 @@ it.effect.each([...@@ -298,6 +299,7 @@ it.effect.each([
298 getSnapshot: () => Effect.die("unused"),299 getSnapshot: () => Effect.die("unused"),
299 getShellSnapshot: () => Effect.die("unused"),300 getShellSnapshot: () => Effect.die("unused"),
300 getDeletedWorktreeThreads: () => Effect.die("unused"),301 getDeletedWorktreeThreads: () => Effect.die("unused"),
302 listThreadsWithPullRequests: () => Effect.die("unused"),
301 getArchivedShellSnapshot: () => Effect.die("unused"),303 getArchivedShellSnapshot: () => Effect.die("unused"),
302 getSnapshotSequence: () => Effect.die("unused"),304 getSnapshotSequence: () => Effect.die("unused"),
303 getCounts: () => Effect.die("unused"),305 getCounts: () => Effect.die("unused"),
...@@ -385,6 +387,7 @@ it.effect(...@@ -385,6 +387,7 @@ it.effect(
385 getSnapshot: () => Effect.die("unused"),387 getSnapshot: () => Effect.die("unused"),
386 getShellSnapshot: () => Effect.die("unused"),388 getShellSnapshot: () => Effect.die("unused"),
387 getDeletedWorktreeThreads: () => Effect.die("unused"),389 getDeletedWorktreeThreads: () => Effect.die("unused"),
390 listThreadsWithPullRequests: () => Effect.die("unused"),
388 getArchivedShellSnapshot: () => Effect.die("unused"),391 getArchivedShellSnapshot: () => Effect.die("unused"),
389 getSnapshotSequence: () => Effect.die("unused"),392 getSnapshotSequence: () => Effect.die("unused"),
390 getCounts: () => Effect.die("unused"),393 getCounts: () => Effect.die("unused"),
...@@ -450,6 +453,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa...@@ -450,6 +453,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
450 getSnapshot: () => Effect.die("unused"),453 getSnapshot: () => Effect.die("unused"),
451 getShellSnapshot: () => Effect.die("unused"),454 getShellSnapshot: () => Effect.die("unused"),
452 getDeletedWorktreeThreads: () => Effect.die("unused"),455 getDeletedWorktreeThreads: () => Effect.die("unused"),
456 listThreadsWithPullRequests: () => Effect.die("unused"),
453 getArchivedShellSnapshot: () => Effect.die("unused"),457 getArchivedShellSnapshot: () => Effect.die("unused"),
454 getSnapshotSequence: () => Effect.die("unused"),458 getSnapshotSequence: () => Effect.die("unused"),
455 getCounts: () => Effect.die("unused"),459 getCounts: () => Effect.die("unused"),