authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-26 01:35:08-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-26 01:35:08-07:00
log1d6f23b51929117d5e69ded8c5c0f8e195d75d4a
tree2d3c0b5e15e7d28fee294df3c46b7039d52e0058
parent294dd123410c77eead2810a052c31fb0cfd3e0c1
signature Signed by PGP key B5690EEEBB952194

perf(server): background sweeps only read threads that can still settle (#13765)

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

6 files changed, 349 insertions(+), 30 deletions(-)

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts+79
...@@ -3643,6 +3643,85 @@ it.effect("reads one sweep thread and its projects like the shell snapshot", ()...@@ -3643,6 +3643,85 @@ it.effect("reads one sweep thread and its projects like the shell snapshot", ()
3643 }).pipe(Effect.provide(layer));3643 }).pipe(Effect.provide(layer));
3644});3644});
36453645
3646it.effect("reads a full sweep from unsettled threads and every project", () => {
3647 const resolved: string[] = [];
3648 const layer = OrchestrationProjectionSnapshotQueryLive.pipe(
3649 Layer.provide(ThreadBackgroundLiveness.layer),
3650 Layer.provide(ThreadPlanProgress.layer),
3651 Layer.provide(
3652 Layer.succeed(RepositoryIdentityResolver.RepositoryIdentityResolver, {
3653 resolve: (root) =>
3654 Effect.sync(() => {
3655 resolved.push(root);
3656 return null;
3657 }),
3658 }),
3659 ),
3660 Layer.provideMerge(SqlitePersistenceMemory),
3661 );
3662 return Effect.gen(function* () {
3663 const sql = yield* SqlClient.SqlClient;
3664 const query = yield* ProjectionSnapshotQuery;
3665 yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at)
3666 VALUES ('p1', 'One', '/one', '[]', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z'),
3667 ('p2', 'Two', '/two', '[]', '2026-09-02T00:00:00Z', '2026-09-02T00:00:00Z'),
3668 ('p3', 'Three', '/three', '[]', '2026-09-03T00:00:00Z', '2026-09-03T00:00:00Z'),
3669 ('p4', 'Four', '/four', '[]', '2026-09-04T00:00:00Z', '2026-09-04T00:00:00Z')`;
3670 yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, branch_pull_request_json, latest_turn_id, created_at, updated_at, archived_at, settled_override, settled_at)
3671 VALUES
3672 ('t-open', 'p1', 'Open', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, 'turn-open', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, NULL, NULL),
3673 ('t-resumed', 'p1', 'Resumed', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, 'active', NULL),
3674 ('t-branch', 'p1', 'Branch', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default',
3675 '{"projectId":"p2","repository":"acme/web","number":8,"url":"https://github.com/acme/web/pull/8"}',
3676 NULL, '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, NULL, NULL),
3677 ('t-settled', 'p3', 'Settled', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, 'turn-settled', '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', NULL, 'settled', '2026-09-03T00:00:00Z'),
3678 ('t-archived', 'p4', 'Archived', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, '2026-09-01T00:00:00Z', '2026-09-01T00:00:00Z', '2026-09-04T00:00:00Z', NULL, NULL)`;
3679 // The open and the settled thread both have a row in each joined table. The
3680 // settled thread's turn and session are the newest rows, so updatedAt shows
3681 // whether those two reads skip it.
3682 yield* sql`INSERT INTO projection_thread_pull_requests (thread_id, host, repository, number, url, source, linked_at)
3683 VALUES ('t-open', 'github.com', 'acme/web', 7, 'https://github.com/acme/web/pull/7', 'agent', '2026-09-02T00:00:00Z'),
3684 ('t-settled', 'github.com', 'acme/web', 9, 'https://github.com/acme/web/pull/9', 'agent', '2026-09-02T00:00:00Z')`;
3685 yield* sql`INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, checkpoint_files_json)
3686 VALUES ('t-open', 'turn-open', 'completed', '2026-09-02T00:00:00Z', '[]'),
3687 ('t-settled', 'turn-settled', 'completed', '2026-09-09T00:00:00Z', '[]')`;
3688 yield* sql`INSERT INTO projection_thread_sessions (thread_id, status, provider_name, active_turn_id, last_error, updated_at)
3689 VALUES ('t-open', 'ready', 'codex', NULL, NULL, '2026-09-02T00:00:00Z'),
3690 ('t-settled', 'stopped', 'codex', NULL, NULL, '2026-09-10T00:00:00Z')`;
3691
3692 const full = yield* query.getShellSnapshot();
3693 // The settled thread's rows must reach the full read, or skipping them proves nothing.
3694 const settled = full.threads.find((thread) => thread.id === ThreadId.make("t-settled"));
3695 assert.strictEqual(settled?.pullRequests[0]?.number, 9);
3696 assert.strictEqual(settled?.latestTurn?.turnId, asTurnId("turn-settled"));
3697 assert.strictEqual(settled?.session?.status, "stopped");
3698 assert.strictEqual(full.updatedAt, "2026-09-10T00:00:00Z");
3699 resolved.length = 0;
3700
3701 const sweep = yield* readSweepSnapshot(query, null);
3702 assert.strictEqual(sweep.snapshotSequence, full.snapshotSequence);
3703 assert.deepStrictEqual(
3704 sweep.threads,
3705 full.threads.filter((thread) => thread.id !== ThreadId.make("t-settled")),
3706 );
3707 assert.deepStrictEqual(
3708 sweep.threads.map((thread) => thread.id),
3709 ["t-branch", "t-open", "t-resumed"],
3710 );
3711 // Like the full read, the sweep resolves every project, so it keeps the
3712 // repository identity cache warm for client connects.
3713 assert.deepStrictEqual(sweep.projects, full.projects);
3714 assert.deepStrictEqual(resolved.toSorted(), ["/four", "/one", "/three", "/two"]);
3715 // A settled thread's link that no longer decodes breaks the full read, but
3716 // not the sweep, which never reads it.
3717 yield* sql`UPDATE projection_thread_pull_requests SET snapshot_json = 'invalid-json' WHERE thread_id = 't-settled'`;
3718 assert.strictEqual((yield* Effect.exit(query.getShellSnapshot()))._tag, "Failure");
3719 const unsettled = yield* query.getShellSnapshot({ unsettledOnly: true });
3720 assert.deepStrictEqual(unsettled.threads, sweep.threads);
3721 assert.strictEqual(unsettled.updatedAt, "2026-09-04T00:00:00Z");
3722 }).pipe(Effect.provide(layer));
3723});
3724
3646projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {3725projectionSnapshotLayer("ProjectionSnapshotQuery activities by kind", (it) => {
3647 it.effect("lists one kind across active threads only, without hydrating the threads", () =>3726 it.effect("lists one kind across active threads only, without hydrating the threads", () =>
3648 Effect.gen(function* () {3727 Effect.gen(function* () {
apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+28-15
...@@ -185,6 +185,7 @@ const EventReplayStatsRowSchema = Schema.Struct({...@@ -185,6 +185,7 @@ const EventReplayStatsRowSchema = Schema.Struct({
185 eventCount: Schema.Number,185 eventCount: Schema.Number,
186 payloadBytes: Schema.Number,186 payloadBytes: Schema.Number,
187});187});
188const ActiveThreadRowsRequest = Schema.Struct({ unsettledOnly: Schema.Boolean });
188const ProjectionThreadSearchRequest = Schema.Struct({189const ProjectionThreadSearchRequest = Schema.Struct({
189 pattern: Schema.String,190 pattern: Schema.String,
190 limit: Schema.Int,191 limit: Schema.Int,
...@@ -600,10 +601,16 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -600,10 +601,16 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
600 `,601 `,
601 });602 });
602603
604 // Background sweeps skip settled threads, the same check PR discovery makes.
605 const unsettledThreadsFilter = (unsettledOnly: boolean) =>
606 unsettledOnly
607 ? sql`AND threads.settled_at IS NULL AND threads.settled_override IS NOT 'settled'`
608 : sql``;
609
603 const listActiveThreadRows = SqlSchema.findAll({610 const listActiveThreadRows = SqlSchema.findAll({
604 Request: Schema.Void,611 Request: ActiveThreadRowsRequest,
605 Result: ProjectionThreadDbRowSchema,612 Result: ProjectionThreadDbRowSchema,
606 execute: () =>613 execute: (request) =>
607 sql`614 sql`
608 SELECT615 SELECT
609 thread_id AS "threadId",616 thread_id AS "threadId",
...@@ -637,9 +644,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -637,9 +644,10 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
637 pending_user_input_count AS "pendingUserInputCount",644 pending_user_input_count AS "pendingUserInputCount",
638 has_actionable_proposed_plan AS "hasActionableProposedPlan",645 has_actionable_proposed_plan AS "hasActionableProposedPlan",
639 deleted_at AS "deletedAt"646 deleted_at AS "deletedAt"
640 FROM projection_threads647 FROM projection_threads threads
641 WHERE deleted_at IS NULL648 WHERE deleted_at IS NULL
642 AND archived_at IS NULL649 AND archived_at IS NULL
650 ${unsettledThreadsFilter(request.unsettledOnly)}
643 ORDER BY project_id ASC, created_at ASC, thread_id ASC651 ORDER BY project_id ASC, created_at ASC, thread_id ASC
644 `,652 `,
645 });653 });
...@@ -779,9 +787,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -779,9 +787,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
779 });787 });
780788
781 const listActiveThreadPullRequestRows = SqlSchema.findAll({789 const listActiveThreadPullRequestRows = SqlSchema.findAll({
782 Request: Schema.Void,790 Request: ActiveThreadRowsRequest,
783 Result: ProjectionThreadPullRequestDbRowSchema,791 Result: ProjectionThreadPullRequestDbRowSchema,
784 execute: () =>792 execute: (request) =>
785 sql`793 sql`
786 SELECT794 SELECT
787 links.thread_id AS "threadId",795 links.thread_id AS "threadId",
...@@ -798,6 +806,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -798,6 +806,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
798 ON threads.thread_id = links.thread_id806 ON threads.thread_id = links.thread_id
799 WHERE threads.deleted_at IS NULL807 WHERE threads.deleted_at IS NULL
800 AND threads.archived_at IS NULL808 AND threads.archived_at IS NULL
809 ${unsettledThreadsFilter(request.unsettledOnly)}
801 ORDER BY links.thread_id ASC, links.linked_at ASC, links.number ASC810 ORDER BY links.thread_id ASC, links.linked_at ASC, links.number ASC
802 `,811 `,
803 });812 });
...@@ -907,9 +916,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -907,9 +916,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
907 });916 });
908917
909 const listActiveThreadSessionRows = SqlSchema.findAll({918 const listActiveThreadSessionRows = SqlSchema.findAll({
910 Request: Schema.Void,919 Request: ActiveThreadRowsRequest,
911 Result: ProjectionThreadSessionDbRowSchema,920 Result: ProjectionThreadSessionDbRowSchema,
912 execute: () =>921 execute: (request) =>
913 sql`922 sql`
914 SELECT923 SELECT
915 sessions.thread_id AS "threadId",924 sessions.thread_id AS "threadId",
...@@ -927,6 +936,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -927,6 +936,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
927 ON threads.thread_id = sessions.thread_id936 ON threads.thread_id = sessions.thread_id
928 WHERE threads.deleted_at IS NULL937 WHERE threads.deleted_at IS NULL
929 AND threads.archived_at IS NULL938 AND threads.archived_at IS NULL
939 ${unsettledThreadsFilter(request.unsettledOnly)}
930 ORDER BY sessions.thread_id ASC940 ORDER BY sessions.thread_id ASC
931 `,941 `,
932 });942 });
...@@ -1001,9 +1011,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -1001,9 +1011,9 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
1001 });1011 });
10021012
1003 const listActiveLatestTurnRows = SqlSchema.findAll({1013 const listActiveLatestTurnRows = SqlSchema.findAll({
1004 Request: Schema.Void,1014 Request: ActiveThreadRowsRequest,
1005 Result: ProjectionLatestTurnDbRowSchema,1015 Result: ProjectionLatestTurnDbRowSchema,
1006 execute: () =>1016 execute: (request) =>
1007 sql`1017 sql`
1008 SELECT1018 SELECT
1009 turns.thread_id AS "threadId",1019 turns.thread_id AS "threadId",
...@@ -1022,6 +1032,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {...@@ -1022,6 +1032,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
1022 WHERE threads.deleted_at IS NULL1032 WHERE threads.deleted_at IS NULL
1023 AND threads.archived_at IS NULL1033 AND threads.archived_at IS NULL
1024 AND threads.latest_turn_id IS NOT NULL1034 AND threads.latest_turn_id IS NOT NULL
1035 ${unsettledThreadsFilter(request.unsettledOnly)}
1025 ORDER BY turns.thread_id ASC1036 ORDER BY turns.thread_id ASC
1026 `,1037 `,
1027 });1038 });
...@@ -2654,8 +2665,9 @@ pending_approval_requests AS (...@@ -2654,8 +2665,9 @@ pending_approval_requests AS (
2654 }),2665 }),
2655 );2666 );
26562667
2657 const getShellSnapshot: ProjectionSnapshotQueryShape["getShellSnapshot"] = () =>2668 const getShellSnapshot: ProjectionSnapshotQueryShape["getShellSnapshot"] = (options) => {
2658 sql2669 const unsettledOnly = options?.unsettledOnly === true;
2670 return sql
2659 .withTransaction(2671 .withTransaction(
2660 Effect.all([2672 Effect.all([
2661 listProjectRows(undefined).pipe(2673 listProjectRows(undefined).pipe(
...@@ -2666,7 +2678,7 @@ pending_approval_requests AS (...@@ -2666,7 +2678,7 @@ pending_approval_requests AS (
2666 ),2678 ),
2667 ),2679 ),
2668 ),2680 ),
2669 listActiveThreadRows(undefined).pipe(2681 listActiveThreadRows({ unsettledOnly }).pipe(
2670 Effect.mapError(2682 Effect.mapError(
2671 toPersistenceSqlOrDecodeError(2683 toPersistenceSqlOrDecodeError(
2672 "ProjectionSnapshotQuery.getShellSnapshot:listThreads:query",2684 "ProjectionSnapshotQuery.getShellSnapshot:listThreads:query",
...@@ -2674,7 +2686,7 @@ pending_approval_requests AS (...@@ -2674,7 +2686,7 @@ pending_approval_requests AS (
2674 ),2686 ),
2675 ),2687 ),
2676 ),2688 ),
2677 listActiveThreadSessionRows(undefined).pipe(2689 listActiveThreadSessionRows({ unsettledOnly }).pipe(
2678 Effect.mapError(2690 Effect.mapError(
2679 toPersistenceSqlOrDecodeError(2691 toPersistenceSqlOrDecodeError(
2680 "ProjectionSnapshotQuery.getShellSnapshot:listThreadSessions:query",2692 "ProjectionSnapshotQuery.getShellSnapshot:listThreadSessions:query",
...@@ -2682,7 +2694,7 @@ pending_approval_requests AS (...@@ -2682,7 +2694,7 @@ pending_approval_requests AS (
2682 ),2694 ),
2683 ),2695 ),
2684 ),2696 ),
2685 listActiveThreadPullRequestRows(undefined).pipe(2697 listActiveThreadPullRequestRows({ unsettledOnly }).pipe(
2686 Effect.mapError(2698 Effect.mapError(
2687 toPersistenceSqlOrDecodeError(2699 toPersistenceSqlOrDecodeError(
2688 "ProjectionSnapshotQuery.getShellSnapshot:listThreadPullRequests:query",2700 "ProjectionSnapshotQuery.getShellSnapshot:listThreadPullRequests:query",
...@@ -2690,7 +2702,7 @@ pending_approval_requests AS (...@@ -2690,7 +2702,7 @@ pending_approval_requests AS (
2690 ),2702 ),
2691 ),2703 ),
2692 ),2704 ),
2693 listActiveLatestTurnRows(undefined).pipe(2705 listActiveLatestTurnRows({ unsettledOnly }).pipe(
2694 Effect.mapError(2706 Effect.mapError(
2695 toPersistenceSqlOrDecodeError(2707 toPersistenceSqlOrDecodeError(
2696 "ProjectionSnapshotQuery.getShellSnapshot:listLatestTurns:query",2708 "ProjectionSnapshotQuery.getShellSnapshot:listLatestTurns:query",
...@@ -2812,6 +2824,7 @@ pending_approval_requests AS (...@@ -2812,6 +2824,7 @@ pending_approval_requests AS (
2812 return toPersistenceSqlError("ProjectionSnapshotQuery.getShellSnapshot:query")(error);2824 return toPersistenceSqlError("ProjectionSnapshotQuery.getShellSnapshot:query")(error);
2813 }),2825 }),
2814 );2826 );
2827 };
28152828
2816 const listThreadsWithPullRequests: ProjectionSnapshotQueryShape["listThreadsWithPullRequests"] =2829 const listThreadsWithPullRequests: ProjectionSnapshotQueryShape["listThreadsWithPullRequests"] =
2817 () =>2830 () =>
apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+8-4
...@@ -120,11 +120,15 @@ export interface ProjectionSnapshotQueryShape {...@@ -120,11 +120,15 @@ export interface ProjectionSnapshotQueryShape {
120 *120 *
121 * Returns only projects and thread shell summaries so clients can bootstrap121 * Returns only projects and thread shell summaries so clients can bootstrap
122 * lightweight navigation state without hydrating every thread body.122 * lightweight navigation state without hydrating every thread body.
123 *
124 * `unsettledOnly` is for background sweeps, not clients. It skips settled
125 * threads and their sessions, PR links, and turns, and its `updatedAt`
126 * ignores those rows. It still resolves every project, which keeps
127 * repository identities cached for client connects.
123 */128 */
124 readonly getShellSnapshot: () => Effect.Effect<129 readonly getShellSnapshot: (options?: {
125 OrchestrationShellSnapshot,130 readonly unsettledOnly?: boolean;
126 ProjectionRepositoryError131 }) => Effect.Effect<OrchestrationShellSnapshot, ProjectionRepositoryError>;
127 >;
128132
129 /**133 /**
130 * Read archived thread shell summaries for the archive page.134 * Read archived thread shell summaries for the archive page.
apps/server/src/orchestration/ThreadPullRequestReactor.test.ts+122-2
...@@ -27,13 +27,21 @@ import * as Queue from "effect/Queue";...@@ -27,13 +27,21 @@ import * as Queue from "effect/Queue";
27import * as Ref from "effect/Ref";27import * as Ref from "effect/Ref";
28import * as Stream from "effect/Stream";28import * as Stream from "effect/Stream";
29import { TestClock } from "effect/testing";29import { TestClock } from "effect/testing";
30import * as SqlClient from "effect/unstable/sql/SqlClient";
3031
31import { GitManager, type GitBranchPullRequest } from "../git/GitManager.ts";32import { GitManager, type GitBranchPullRequest } from "../git/GitManager.ts";
33import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
32import { PullRequestService } from "../pullRequest/PullRequestService.ts";34import { PullRequestService } from "../pullRequest/PullRequestService.ts";
33import { RepositoryIdentityResolver } from "../project/RepositoryIdentityResolver.ts";35import { RepositoryIdentityResolver } from "../project/RepositoryIdentityResolver.ts";
34import { ServerActivation } from "../serverActivation.ts";36import { ServerActivation } from "../serverActivation.ts";
37import { OrchestrationProjectionSnapshotQueryLive } from "./Layers/ProjectionSnapshotQuery.ts";
35import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";38import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";
36import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";39import {
40 ProjectionSnapshotQuery,
41 type ProjectionSnapshotQueryShape,
42} from "./Services/ProjectionSnapshotQuery.ts";
43import * as ThreadBackgroundLiveness from "./ThreadBackgroundLiveness.ts";
44import * as ThreadPlanProgress from "./ThreadPlanProgress.ts";
37import * as ThreadPullRequestReactor from "./ThreadPullRequestReactor.ts";45import * as ThreadPullRequestReactor from "./ThreadPullRequestReactor.ts";
3846
39const NOW = "2026-09-01T12:00:00.000Z";47const NOW = "2026-09-01T12:00:00.000Z";
...@@ -135,6 +143,8 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options...@@ -135,6 +143,8 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options
135 readonly existingWorktrees?: ReadonlyArray<string>;143 readonly existingWorktrees?: ReadonlyArray<string>;
136 readonly project?: OrchestrationProjectShell;144 readonly project?: OrchestrationProjectShell;
137 readonly resolveRepositoryIdentity?: RepositoryIdentityResolver["Service"]["resolve"];145 readonly resolveRepositoryIdentity?: RepositoryIdentityResolver["Service"]["resolve"];
146 /** Serve full sweep reads from this instead of `threads`. */
147 readonly getShellSnapshot?: ProjectionSnapshotQueryShape["getShellSnapshot"];
138}) {148}) {
139 const activation = yield* Deferred.make<void>();149 const activation = yield* Deferred.make<void>();
140 const snapshots = yield* Ref.make<OrchestrationShellSnapshot>({150 const snapshots = yield* Ref.make<OrchestrationShellSnapshot>({
...@@ -154,7 +164,10 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options...@@ -154,7 +164,10 @@ const makeHarness = Effect.fn("makeThreadPullRequestHarness")(function* (options
154 let uuid = 0;164 let uuid = 0;
155 const dependencies = Layer.mergeAll(165 const dependencies = Layer.mergeAll(
156 Layer.mock(ProjectionSnapshotQuery)({166 Layer.mock(ProjectionSnapshotQuery)({
157 getShellSnapshot: () => Ref.get(snapshots).pipe(Effect.tap(() => Queue.offer(reads, null))),167 getShellSnapshot: (readOptions) =>
168 (options.getShellSnapshot?.(readOptions) ?? Ref.get(snapshots)).pipe(
169 Effect.tap(() => Queue.offer(reads, null)),
170 ),
158 getSnapshotSequence: () =>171 getSnapshotSequence: () =>
159 Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))),172 Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))),
160 getThreadShellById: (threadId) =>173 getThreadShellById: (threadId) =>
...@@ -644,6 +657,113 @@ describe("ThreadPullRequestReactor", () => {...@@ -644,6 +657,113 @@ describe("ThreadPullRequestReactor", () => {
644 ),657 ),
645 );658 );
646659
660 it.effect("discovers the same PRs from the unsettled read as from the full read", () =>
661 Effect.gen(function* () {
662 const sql = yield* SqlClient.SqlClient;
663 const query = yield* ProjectionSnapshotQuery;
664 yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at)
665 VALUES ('project', 'Project', '/workspace/project', '[]', ${NOW}, ${NOW}),
666 ('dormant', 'Dormant', '/workspace/dormant', '[]', ${NOW}, ${NOW})`;
667 yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, branch, branch_pull_request_json, created_at, updated_at, archived_at, settled_override, settled_at)
668 VALUES
669 ('open', 'project', 'Open', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'open', NULL, ${NOW}, ${NOW}, NULL, NULL, NULL),
670 ('resumed', 'project', 'Resumed', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'resumed', NULL, ${NOW}, ${NOW}, NULL, 'active', NULL),
671 ('linked', 'project', 'Linked', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'linked', '{"projectId":"project","repository":"owner/repository","number":3,"url":"https://github.com/owner/repository/pull/3"}', ${NOW}, ${NOW}, NULL, NULL, NULL),
672 ('settled', 'project', 'Settled', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'settled', '{"projectId":"project","repository":"owner/repository","number":4,"url":"https://github.com/owner/repository/pull/4"}', ${NOW}, ${NOW}, NULL, 'settled', ${NOW}),
673 ('backfill', 'project', 'Backfill', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'backfill', NULL, ${NOW}, ${NOW}, NULL, 'settled', ${NOW}),
674 ('imported', 'dormant', 'Imported', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, ${NOW}, ${NOW}, NULL, 'settled', ${NOW}),
675 ('archived', 'project', 'Archived', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'archived', NULL, ${NOW}, ${NOW}, ${NOW}, NULL, NULL)`;
676 const numbers = new Map([
677 ["open", 1],
678 ["resumed", 2],
679 ["linked", 3],
680 ["settled", 4],
681 ["backfill", 5],
682 ["archived", 6],
683 ]);
684
685 // Startup, then two periodic passes. The startup backfill lookup fails,
686 // so the first periodic pass retries it from the full read.
687 const discover = (read: ProjectionSnapshotQueryShape["getShellSnapshot"]) =>
688 Effect.gen(function* () {
689 const online = yield* Ref.make(false);
690 const reads: Array<ReadonlyArray<string>> = [];
691 const fixture = yield* makeHarness({
692 threads: [],
693 getShellSnapshot: (options) =>
694 read(options).pipe(
695 Effect.tap((snapshot) =>
696 Effect.sync(() => reads.push(snapshot.threads.map(({ id }) => id).toSorted())),
697 ),
698 ),
699 branchPullRequest: ({ cwd, branch }) =>
700 Effect.gen(function* () {
701 if (branch === "backfill" && !(yield* Ref.get(online))) {
702 return yield* new GitManagerError({
703 operation: "branchPullRequest",
704 cwd,
705 detail: "Offline",
706 });
707 }
708 const number = numbers.get(branch);
709 return number === undefined ? null : branchPullRequest(number);
710 }),
711 });
712 return yield* Effect.gen(function* () {
713 const reactor = yield* fixture.start();
714 const startupCalls = (yield* Ref.get(fixture.branchCalls)).length;
715 const startupCommands = (yield* Ref.get(fixture.commands)).length;
716 yield* Ref.set(online, true);
717 for (let pass = 0; pass < 2; pass++) {
718 yield* TestClock.adjust("1 minute");
719 yield* Queue.take(fixture.reads);
720 yield* reactor.drain;
721 }
722 return {
723 reads,
724 branchCalls: (yield* Ref.get(fixture.branchCalls))
725 .slice(startupCalls)
726 .map(({ branch }) => branch)
727 .toSorted(),
728 commands: (yield* Ref.get(fixture.commands))
729 .slice(startupCommands)
730 .map(
731 ({ threadId, branchPullRequest }) => `${threadId} ${branchPullRequest?.number}`,
732 )
733 .toSorted(),
734 };
735 }).pipe(Effect.provide(fixture.layer));
736 }).pipe(Effect.scoped);
737
738 const { reads: unsettledReads, ...unsettled } = yield* discover(query.getShellSnapshot);
739 const { reads: fullReads, ...full } = yield* discover(() => query.getShellSnapshot());
740 expect(unsettled).toEqual(full);
741 expect(unsettled.commands).toEqual([
742 "backfill 5",
743 "open 1",
744 "open 1",
745 "resumed 2",
746 "resumed 2",
747 ]);
748 // The last pass has no backfill left, so it reads no settled thread.
749 expect(fullReads.at(-1)).toContain("imported");
750 expect(unsettledReads.at(-1)).toEqual(["linked", "open", "resumed"]);
751 }).pipe(
752 Effect.provide(
753 OrchestrationProjectionSnapshotQueryLive.pipe(
754 Layer.provide(ThreadBackgroundLiveness.layer),
755 Layer.provide(ThreadPlanProgress.layer),
756 Layer.provide(
757 Layer.succeed(RepositoryIdentityResolver, {
758 resolve: () => Effect.succeed(project.repositoryIdentity),
759 }),
760 ),
761 Layer.provideMerge(SqlitePersistenceMemory),
762 ),
763 ),
764 ),
765 );
766
647 it.effect("matches Azure SSH projects to HTTPS PRs with the provider repository selector", () =>767 it.effect("matches Azure SSH projects to HTTPS PRs with the provider repository selector", () =>
648 Effect.scoped(768 Effect.scoped(
649 Effect.gen(function* () {769 Effect.gen(function* () {
apps/server/src/orchestration/ThreadPullRequestReactor.ts+16-6
...@@ -74,7 +74,9 @@ export function pullRequestMatchesProject(...@@ -74,7 +74,9 @@ export function pullRequestMatchesProject(
7474
75/**75/**
76 * Read the shell state for a discovery or settlement sweep. A sweep for one76 * 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.77 * thread reads that thread and the projects it names, not every thread. A
78 * sweep over all threads reads only unsettled threads, since both sweeps skip
79 * settled ones. Discovery's backfill does its own full read.
78 */80 */
79export const readSweepSnapshot = (81export const readSweepSnapshot = (
80 snapshots: ProjectionSnapshotQuery.ProjectionSnapshotQueryShape,82 snapshots: ProjectionSnapshotQuery.ProjectionSnapshotQueryShape,
...@@ -84,7 +86,7 @@ export const readSweepSnapshot = (...@@ -84,7 +86,7 @@ export const readSweepSnapshot = (
84 ProjectionRepositoryError86 ProjectionRepositoryError
85> =>87> =>
86 threadId === null88 threadId === null
87 ? snapshots.getShellSnapshot()89 ? snapshots.getShellSnapshot({ unsettledOnly: true })
88 : Effect.gen(function* () {90 : Effect.gen(function* () {
89 // Read the sequence first. The thread is then at least this new, so a91 // 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.92 // command guarded by the sequence is rejected rather than missing a change.
...@@ -129,7 +131,11 @@ export const make = Effect.gen(function* () {...@@ -129,7 +131,11 @@ export const make = Effect.gen(function* () {
129 const synchronize = Effect.fn("ThreadPullRequestReactor.synchronize")(function* (131 const synchronize = Effect.fn("ThreadPullRequestReactor.synchronize")(function* (
130 request: RefreshRequest,132 request: RefreshRequest,
131 ) {133 ) {
132 const snapshot = yield* readSweepSnapshot(snapshots, request.threadId);134 // Backfill looks up settled threads, so its passes read every thread.
135 const snapshot =
136 request.threadId === null && (request.backfill || pendingBackfill.size > 0)
137 ? yield* snapshots.getShellSnapshot()
138 : yield* readSweepSnapshot(snapshots, request.threadId);
133 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));139 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
134 if (request.backfill) {140 if (request.backfill) {
135 for (const thread of snapshot.threads) {141 for (const thread of snapshot.threads) {
...@@ -141,11 +147,15 @@ export const make = Effect.gen(function* () {...@@ -141,11 +147,15 @@ export const make = Effect.gen(function* () {
141 }147 }
142 }148 }
143 }149 }
144 // A single-thread read only shows whether its own thread is gone.150 // A single-thread read only shows whether its own thread is gone. A thread
145 const threadIds = new Set(snapshot.threads.map((thread) => thread.id));151 // with no branch has nothing to look up, and its entry would keep every
152 // periodic pass on the full read.
153 const branchThreadIds = new Set(
154 snapshot.threads.filter((thread) => thread.branch !== null).map((thread) => thread.id),
155 );
146 const checkedIds = request.threadId === null ? pendingBackfill.keys() : [request.threadId];156 const checkedIds = request.threadId === null ? pendingBackfill.keys() : [request.threadId];
147 for (const threadId of checkedIds) {157 for (const threadId of checkedIds) {
148 if (!threadIds.has(threadId)) pendingBackfill.delete(threadId);158 if (!branchThreadIds.has(threadId)) pendingBackfill.delete(threadId);
149 }159 }
150 const threads = snapshot.threads.filter(160 const threads = snapshot.threads.filter(
151 (thread) =>161 (thread) =>
apps/server/src/orchestration/ThreadSettlementReactor.test.ts+96-3
...@@ -30,8 +30,11 @@ import * as Queue from "effect/Queue";...@@ -30,8 +30,11 @@ import * as Queue from "effect/Queue";
30import * as Ref from "effect/Ref";30import * as Ref from "effect/Ref";
31import * as Stream from "effect/Stream";31import * as Stream from "effect/Stream";
32import { TestClock } from "effect/testing";32import { TestClock } from "effect/testing";
33import * as SqlClient from "effect/unstable/sql/SqlClient";
3334
34import { GitManager, type GitBranchPullRequest } from "../git/GitManager.ts";35import { GitManager, type GitBranchPullRequest } from "../git/GitManager.ts";
36import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
37import { RepositoryIdentityResolver } from "../project/RepositoryIdentityResolver.ts";
35import {38import {
36 PullRequestService,39 PullRequestService,
37 type PullRequestMergeEvent,40 type PullRequestMergeEvent,
...@@ -39,11 +42,17 @@ import {...@@ -39,11 +42,17 @@ import {
39import { ServerActivation } from "../serverActivation.ts";42import { ServerActivation } from "../serverActivation.ts";
40import { ServerSettingsService } from "../serverSettings.ts";43import { ServerSettingsService } from "../serverSettings.ts";
41import { OrchestrationCommandInvariantError } from "./Errors.ts";44import { OrchestrationCommandInvariantError } from "./Errors.ts";
45import { OrchestrationProjectionSnapshotQueryLive } from "./Layers/ProjectionSnapshotQuery.ts";
42import {46import {
43 OrchestrationEngineService,47 OrchestrationEngineService,
44 type OrchestrationEngineShape,48 type OrchestrationEngineShape,
45} from "./Services/OrchestrationEngine.ts";49} from "./Services/OrchestrationEngine.ts";
46import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";50import {
51 ProjectionSnapshotQuery,
52 type ProjectionSnapshotQueryShape,
53} from "./Services/ProjectionSnapshotQuery.ts";
54import * as ThreadBackgroundLiveness from "./ThreadBackgroundLiveness.ts";
55import * as ThreadPlanProgress from "./ThreadPlanProgress.ts";
47import * as ThreadSettlementReactor from "./ThreadSettlementReactor.ts";56import * as ThreadSettlementReactor from "./ThreadSettlementReactor.ts";
48import * as NodeServices from "@effect/platform-node/NodeServices";57import * as NodeServices from "@effect/platform-node/NodeServices";
49import * as Path from "effect/Path";58import * as Path from "effect/Path";
...@@ -169,6 +178,8 @@ function makeBranchPullRequest(...@@ -169,6 +178,8 @@ function makeBranchPullRequest(
169178
170interface HarnessOptions {179interface HarnessOptions {
171 readonly snapshot: OrchestrationShellSnapshot;180 readonly snapshot: OrchestrationShellSnapshot;
181 /** Serve full sweep reads from this instead of `snapshot`. */
182 readonly getShellSnapshot?: ProjectionSnapshotQueryShape["getShellSnapshot"];
172 readonly settings?: ServerSettings;183 readonly settings?: ServerSettings;
173 readonly branchPullRequest?: GitManager["Service"]["branchPullRequest"];184 readonly branchPullRequest?: GitManager["Service"]["branchPullRequest"];
174 readonly pullRequestSummary?: PullRequestService["Service"]["summary"];185 readonly pullRequestSummary?: PullRequestService["Service"]["summary"];
...@@ -256,10 +267,10 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -256,10 +267,10 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
256267
257 const dependencies = Layer.mergeAll(268 const dependencies = Layer.mergeAll(
258 Layer.mock(ProjectionSnapshotQuery)({269 Layer.mock(ProjectionSnapshotQuery)({
259 getShellSnapshot: () =>270 getShellSnapshot: (readOptions) =>
260 Ref.update(snapshotReadCount, (count) => count + 1).pipe(271 Ref.update(snapshotReadCount, (count) => count + 1).pipe(
261 Effect.andThen(Queue.offer(snapshotReads, null)),272 Effect.andThen(Queue.offer(snapshotReads, null)),
262 Effect.andThen(Ref.get(snapshots)),273 Effect.andThen(options.getShellSnapshot?.(readOptions) ?? Ref.get(snapshots)),
263 ),274 ),
264 getSnapshotSequence: () =>275 getSnapshotSequence: () =>
265 Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))),276 Ref.get(snapshots).pipe(Effect.map(({ snapshotSequence }) => ({ snapshotSequence }))),
...@@ -1437,6 +1448,88 @@ describe("ThreadSettlementReactor", () => {...@@ -1437,6 +1448,88 @@ describe("ThreadSettlementReactor", () => {
1437 }),1448 }),
1438 ),1449 ),
1439 );1450 );
1451
1452 it.effect("settles the same threads from the unsettled read as from the full read", () =>
1453 Effect.gen(function* () {
1454 yield* TestClock.setTime(Date.parse(NOW));
1455 const sql = yield* SqlClient.SqlClient;
1456 const query = yield* ProjectionSnapshotQuery;
1457 yield* sql`INSERT INTO projection_projects (project_id, title, workspace_root, scripts_json, created_at, updated_at)
1458 VALUES ('settlement-project', 'Project', '/workspace/project', '[]', ${NOW}, ${NOW}),
1459 ('linked-settlement-project', 'Linked', '/workspace/linked', '[]', ${NOW}, ${NOW}),
1460 ('dormant-project', 'Dormant', '/workspace/dormant', '[]', ${NOW}, ${NOW})`;
1461 yield* sql`INSERT INTO projection_threads (thread_id, project_id, title, model_selection_json, runtime_mode, interaction_mode, branch, branch_pull_request_json, latest_user_message_at, created_at, updated_at, archived_at, settled_override, settled_at)
1462 VALUES
1463 ('idle', 'settlement-project', 'Idle', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, '2026-08-20T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, NULL, NULL, NULL),
1464 ('merged', 'settlement-project', 'Merged', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'feature',
1465 '{"projectId":"linked-settlement-project","repository":"owner/repository","number":42,"url":"https://example.test/owner/repository/pull/42"}',
1466 '2026-08-27T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, NULL, NULL, NULL),
1467 ('linked', 'settlement-project', 'Linked', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, '2026-08-27T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, NULL, NULL, NULL),
1468 ('open', 'settlement-project', 'Open', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'open-feature', NULL, '2026-08-27T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, NULL, NULL, NULL),
1469 ('resumed', 'settlement-project', 'Resumed', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, '2026-08-20T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, NULL, 'active', NULL),
1470 ('settled', 'dormant-project', 'Settled', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', 'done', NULL, '2026-08-20T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, NULL, 'settled', '2026-08-21T00:00:00.000Z'),
1471 ('archived', 'settlement-project', 'Archived', '{"provider":"codex","model":"gpt-5"}', 'full-access', 'default', NULL, NULL, '2026-08-20T00:00:00.000Z', '2026-08-01T00:00:00.000Z', ${NOW}, ${NOW}, NULL, NULL)`;
1472 yield* sql`INSERT INTO projection_thread_pull_requests (thread_id, host, repository, number, url, source, linked_at, snapshot_json)
1473 VALUES ('linked', 'example.test', 'owner/repository', 7, 'https://example.test/owner/repository/pull/7', 'manual', ${NOW},
1474 '{"state":"merged","title":"Review","headBranch":"linked","baseBranch":"main","isDraft":false,"updatedAt":"2026-08-28T12:00:00.000Z","syncedAt":"2026-08-28T12:00:00.000Z","mergedAt":"2026-08-28T12:00:00.000Z","closedAt":null}')`;
1475
1476 const sweep = (read: ProjectionSnapshotQueryShape["getShellSnapshot"]) =>
1477 Effect.gen(function* () {
1478 const readThreadIds: Array<string> = [];
1479 const fixture = yield* makeHarness({
1480 snapshot: makeSnapshot([]),
1481 getShellSnapshot: (options) =>
1482 read(options).pipe(
1483 Effect.tap((snapshot) =>
1484 Effect.sync(() => readThreadIds.push(...snapshot.threads.map(({ id }) => id))),
1485 ),
1486 ),
1487 branchPullRequest: ({ branch }) =>
1488 Effect.succeed(branch === "open-feature" ? makeBranchPullRequest("open") : null),
1489 pullRequestSummary: (input) =>
1490 Effect.succeed(makePullRequestSummary({ ...input, state: "merged" })),
1491 });
1492 return yield* Effect.gen(function* () {
1493 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
1494 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
1495 return {
1496 readThreadIds: readThreadIds.toSorted(),
1497 commands: (yield* Ref.get(fixture.commands))
1498 .map(({ threadId, settledAt }) => `${threadId} ${settledAt}`)
1499 .toSorted(),
1500 branchCalls: (yield* Ref.get(fixture.branchCalls))
1501 .map(({ branch }) => branch)
1502 .toSorted(),
1503 summaryCalls: yield* Ref.get(fixture.summaryCalls),
1504 };
1505 }).pipe(Effect.provide(fixture.layer));
1506 }).pipe(Effect.scoped);
1507
1508 const { readThreadIds: unsettledReads, ...unsettled } = yield* sweep(query.getShellSnapshot);
1509 const { readThreadIds: fullReads, ...full } = yield* sweep(() => query.getShellSnapshot());
1510 assert.deepStrictEqual(unsettled, full);
1511 // A settle for inactivity, for a synced merged link, and for a saved
1512 // branch PR whose project only that PR names.
1513 assert.deepStrictEqual(unsettled.commands, [
1514 "idle 2026-08-20T00:00:00.000Z",
1515 "linked 2026-08-27T00:00:00.000Z",
1516 "merged 2026-08-27T00:00:00.000Z",
1517 ]);
1518 assert.deepStrictEqual(fullReads, [...unsettledReads, "settled"].toSorted());
1519 assert.deepStrictEqual(unsettledReads, ["idle", "linked", "merged", "open", "resumed"]);
1520 }).pipe(
1521 Effect.provide(
1522 OrchestrationProjectionSnapshotQueryLive.pipe(
1523 Layer.provide(ThreadBackgroundLiveness.layer),
1524 Layer.provide(ThreadPlanProgress.layer),
1525 Layer.provide(
1526 Layer.succeed(RepositoryIdentityResolver, { resolve: () => Effect.succeed(null) }),
1527 ),
1528 Layer.provideMerge(SqlitePersistenceMemory),
1529 ),
1530 ),
1531 ),
1532 );
1440});1533});
14411534
1442describe("storage cleanup", () => {1535describe("storage cleanup", () => {