authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-06 02:15:45-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-06 02:15:45-07:00
logf66cfe2219e3f255f573f6430f9c989c58ed517d
tree5858a9b5112542ae86c4e7539238a477a5a2c3ca
parent4c7cd17a832269ef17edcf99e987ac25ae1b640a
signature Signed by PGP key B5690EEEBB952194

fix(server): settle inactive threads without a PR lookup (#10103)

2 files changed, 199 insertions(+), 47 deletions(-)

apps/server/src/orchestration/ThreadSettlementReactor.test.ts+144-9
...@@ -305,7 +305,10 @@ describe("ThreadSettlementReactor", () => {...@@ -305,7 +305,10 @@ describe("ThreadSettlementReactor", () => {
305 snapshot: makeSnapshot(305 snapshot: makeSnapshot(
306 [306 [
307 makeThread("inactive", { branch: "inactive-feature" }),307 makeThread("inactive", { branch: "inactive-feature" }),
308 makeThread("closed-pr", { linkedPullRequest }),308 makeThread("closed-pr", {
309 linkedPullRequest,
310 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
311 }),
309 ...skipped,312 ...skipped,
310 ],313 ],
311 [makeProject(), makeProject(LINKED_PROJECT_ID, "/workspace/linked")],314 [makeProject(), makeProject(LINKED_PROJECT_ID, "/workspace/linked")],
...@@ -337,7 +340,7 @@ describe("ThreadSettlementReactor", () => {...@@ -337,7 +340,7 @@ describe("ThreadSettlementReactor", () => {
337 {340 {
338 threadId: ThreadId.make("closed-pr"),341 threadId: ThreadId.make("closed-pr"),
339 snapshotSequence: 1,342 snapshotSequence: 1,
340 settledAt: "2026-08-20T00:00:00.000Z",343 settledAt: "2026-08-27T00:00:00.000Z",
341 },344 },
342 {345 {
343 threadId: ThreadId.make("inactive"),346 threadId: ThreadId.make("inactive"),
...@@ -346,9 +349,7 @@ describe("ThreadSettlementReactor", () => {...@@ -346,9 +349,7 @@ describe("ThreadSettlementReactor", () => {
346 },349 },
347 ],350 ],
348 );351 );
349 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), [352 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), []);
350 { cwd: "/workspace/project", branch: "inactive-feature" },
351 ]);
352 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), [353 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), [
353 { projectId: LINKED_PROJECT_ID, repository: "owner/repository", number: 42 },354 { projectId: LINKED_PROJECT_ID, repository: "owner/repository", number: 42 },
354 ]);355 ]);
...@@ -589,7 +590,12 @@ describe("ThreadSettlementReactor", () => {...@@ -589,7 +590,12 @@ describe("ThreadSettlementReactor", () => {
589 const releaseLaterLookup = yield* Deferred.make<void>();590 const releaseLaterLookup = yield* Deferred.make<void>();
590 const lookupCount = yield* Ref.make(0);591 const lookupCount = yield* Ref.make(0);
591 const fixture = yield* makeHarness({592 const fixture = yield* makeHarness({
592 snapshot: makeSnapshot([makeThread("settings-thread", { branch: "saved-feature" })]),593 snapshot: makeSnapshot([
594 makeThread("settings-thread", {
595 branch: "saved-feature",
596 latestUserMessageAt: "2026-08-28T00:00:00.000Z",
597 }),
598 ]),
593 settings: {599 settings: {
594 ...DEFAULT_SERVER_SETTINGS,600 ...DEFAULT_SERVER_SETTINGS,
595 sidebarAutoSettleAfterDays: null,601 sidebarAutoSettleAfterDays: null,
...@@ -661,6 +667,7 @@ describe("ThreadSettlementReactor", () => {...@@ -661,6 +667,7 @@ describe("ThreadSettlementReactor", () => {
661 snapshot: makeSnapshot(667 snapshot: makeSnapshot(
662 [668 [
663 makeThread("lookup-failed", {669 makeThread("lookup-failed", {
670 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
664 linkedPullRequest: {671 linkedPullRequest: {
665 projectId: LINKED_PROJECT_ID,672 projectId: LINKED_PROJECT_ID,
666 repository: "owner/repository",673 repository: "owner/repository",
...@@ -695,6 +702,121 @@ describe("ThreadSettlementReactor", () => {...@@ -695,6 +702,121 @@ describe("ThreadSettlementReactor", () => {
695 ),702 ),
696 );703 );
697704
705 it.effect("settles inactive linked and branch threads without reading an unavailable host", () =>
706 Effect.scoped(
707 Effect.gen(function* () {
708 yield* TestClock.setTime(Date.parse(NOW));
709 const fixture = yield* makeHarness({
710 snapshot: makeSnapshot([
711 makeThread("inactive-linked", {
712 linkedPullRequest: {
713 projectId: PROJECT_ID,
714 repository: "owner/repository",
715 number: 42,
716 url: "https://example.test/owner/repository/pull/42",
717 },
718 }),
719 makeThread("inactive-branch", {
720 branch: "saved-feature",
721 latestUserMessageAt: "2026-08-21T00:00:00.000Z",
722 }),
723 ]),
724 branchPullRequest: () => Effect.die(new Error("host unavailable")),
725 pullRequestSummary: () =>
726 Effect.fail(
727 new PullRequestOperationError({
728 operation: "summary",
729 detail: "host unavailable",
730 }),
731 ),
732 });
733
734 yield* Effect.gen(function* () {
735 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
736 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
737
738 assert.deepStrictEqual(
739 (yield* Ref.get(fixture.commands))
740 .map(({ threadId, snapshotSequence, settledAt }) => ({
741 threadId,
742 snapshotSequence,
743 settledAt,
744 }))
745 .sort((left, right) => left.threadId.localeCompare(right.threadId)),
746 [
747 {
748 threadId: ThreadId.make("inactive-branch"),
749 snapshotSequence: 1,
750 settledAt: "2026-08-21T00:00:00.000Z",
751 },
752 {
753 threadId: ThreadId.make("inactive-linked"),
754 snapshotSequence: 1,
755 settledAt: "2026-08-20T00:00:00.000Z",
756 },
757 ],
758 );
759 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []);
760 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), []);
761 }).pipe(Effect.provide(fixture.layer));
762 }),
763 ),
764 );
765
766 it.effect("settles an inactive thread before its shared pull request lookup completes", () =>
767 Effect.scoped(
768 Effect.gen(function* () {
769 yield* TestClock.setTime(Date.parse(NOW));
770 const lookupStarted = yield* Deferred.make<void>();
771 const releaseLookup = yield* Deferred.make<void>();
772 const linkedPullRequest = {
773 projectId: PROJECT_ID,
774 repository: "owner/repository",
775 number: 42,
776 url: "https://example.test/owner/repository/pull/42",
777 } as const;
778 const fixture = yield* makeHarness({
779 snapshot: makeSnapshot([
780 makeThread("recent-linked", {
781 linkedPullRequest,
782 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
783 }),
784 makeThread("inactive-linked", { linkedPullRequest }),
785 ]),
786 pullRequestSummary: () =>
787 Deferred.succeed(lookupStarted, undefined).pipe(
788 Effect.andThen(Deferred.await(releaseLookup)),
789 Effect.andThen(
790 Effect.fail(
791 new PullRequestOperationError({
792 operation: "summary",
793 detail: "host unavailable",
794 }),
795 ),
796 ),
797 ),
798 });
799
800 yield* Effect.gen(function* () {
801 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
802 yield* reactor.start();
803 yield* Deferred.succeed(fixture.activation, undefined);
804 yield* Deferred.await(lookupStarted);
805
806 assert.deepStrictEqual(
807 (yield* Ref.get(fixture.commands)).map(({ threadId }) => threadId),
808 [ThreadId.make("inactive-linked")],
809 );
810
811 yield* Deferred.succeed(releaseLookup, undefined);
812 yield* reactor.drain;
813 assert.strictEqual((yield* Ref.get(fixture.commands)).length, 1);
814 assert.strictEqual((yield* Ref.get(fixture.summaryCalls)).length, 1);
815 }).pipe(Effect.provide(fixture.layer));
816 }),
817 ),
818 );
819
698 it.effect("keeps threads active when their pull request project is unavailable", () =>820 it.effect("keeps threads active when their pull request project is unavailable", () =>
699 Effect.scoped(821 Effect.scoped(
700 Effect.gen(function* () {822 Effect.gen(function* () {
...@@ -712,7 +834,10 @@ describe("ThreadSettlementReactor", () => {...@@ -712,7 +834,10 @@ describe("ThreadSettlementReactor", () => {
712 latestUserMessageAt: "2026-08-27T00:00:00.000Z",834 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
713 linkedPullRequest,835 linkedPullRequest,
714 }),836 }),
715 makeThread("missing-branch-project", { branch: "saved-feature" }),837 makeThread("missing-branch-project", {
838 branch: "saved-feature",
839 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
840 }),
716 ],841 ],
717 [makeProject(LINKED_PROJECT_ID, "/workspace/linked")],842 [makeProject(LINKED_PROJECT_ID, "/workspace/linked")],
718 ),843 ),
...@@ -750,13 +875,21 @@ describe("ThreadSettlementReactor", () => {...@@ -750,13 +875,21 @@ describe("ThreadSettlementReactor", () => {
750 makeThread("branch-one", {875 makeThread("branch-one", {
751 branch: "saved-feature",876 branch: "saved-feature",
752 worktreePath: "/deleted/worktree-one",877 worktreePath: "/deleted/worktree-one",
878 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
753 }),879 }),
754 makeThread("branch-two", {880 makeThread("branch-two", {
755 branch: "saved-feature",881 branch: "saved-feature",
756 worktreePath: "/deleted/worktree-two",882 worktreePath: "/deleted/worktree-two",
883 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
884 }),
885 makeThread("linked-one", {
886 linkedPullRequest,
887 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
888 }),
889 makeThread("linked-two", {
890 linkedPullRequest,
891 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
757 }),892 }),
758 makeThread("linked-one", { linkedPullRequest }),
759 makeThread("linked-two", { linkedPullRequest }),
760 ],893 ],
761 [894 [
762 makeProject(PROJECT_ID, "/workspace/project-root"),895 makeProject(PROJECT_ID, "/workspace/project-root"),
...@@ -803,10 +936,12 @@ describe("ThreadSettlementReactor", () => {...@@ -803,10 +936,12 @@ describe("ThreadSettlementReactor", () => {
803 makeThread("live-worktree", {936 makeThread("live-worktree", {
804 branch: "feature/live",937 branch: "feature/live",
805 worktreePath: "/workspace/project-root/.worktrees/live",938 worktreePath: "/workspace/project-root/.worktrees/live",
939 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
806 }),940 }),
807 makeThread("deleted-worktree", {941 makeThread("deleted-worktree", {
808 branch: "feature/deleted",942 branch: "feature/deleted",
809 worktreePath: "/workspace/project-root/.worktrees/deleted",943 worktreePath: "/workspace/project-root/.worktrees/deleted",
944 latestUserMessageAt: "2026-08-27T00:00:00.000Z",
810 }),945 }),
811 ],946 ],
812 [makeProject(PROJECT_ID, "/workspace/project-root")],947 [makeProject(PROJECT_ID, "/workspace/project-root")],
apps/server/src/orchestration/ThreadSettlementReactor.ts+55-38
...@@ -51,10 +51,60 @@ export const make = Effect.gen(function* () {...@@ -51,10 +51,60 @@ export const make = Effect.gen(function* () {
51 // their branch lookup, which would otherwise wait for the next minute's51 // their branch lookup, which would otherwise wait for the next minute's
52 // sweep on a possibly stale cached answer.52 // sweep on a possibly stale cached answer.
53 const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));53 const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));
54
55 // Return the thread when it still needs a pull request decision. A rejected
56 // dispatch skips it for this snapshot instead of retrying through a lookup.
57 const settleThread = Effect.fn("ThreadSettlementReactor.settleThread")(
58 function* (thread: (typeof candidates)[number], pullRequest: SettlementPullRequest | null) {
59 const settings = yield* settingsService.getSettings;
60 const decisionNow = DateTime.formatIso(yield* DateTime.now);
61 const settledAt = resolveAutoSettlementAt({
62 thread,
63 pullRequest,
64 now: decisionNow,
65 autoSettleAfterDays: settings.sidebarAutoSettleAfterDays,
66 autoSettleOnMerge: settings.sidebarAutoSettleOnMerge,
67 });
68 if (settledAt === null) {
69 return thread;
70 }
71 const uuid = yield* crypto.randomUUIDv4;
72 yield* engine.dispatch({
73 type: "thread.auto-settle",
74 commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),
75 threadId: thread.id,
76 snapshotSequence: snapshot.snapshotSequence,
77 settledAt,
78 });
79 return null;
80 },
81 (effect, thread) =>
82 effect.pipe(
83 Effect.catchCause((cause) =>
84 Cause.hasInterruptsOnly(cause)
85 ? Effect.failCause(cause)
86 : Effect.logWarning("automatic thread settlement skipped", {
87 threadId: thread.id,
88 cause: Cause.pretty(cause),
89 }).pipe(Effect.as(null)),
90 ),
91 ),
92 );
93
94 // Inactivity needs no host state. Finish these decisions before any lookup
95 // can fail or wait on the network, including lookups shared by recent threads.
96 const lookupCandidates = (yield* Effect.forEach(
97 candidates,
98 (thread) => settleThread(thread, null),
99 {
100 concurrency: 8,
101 },
102 )).filter((thread) => thread !== null);
103
54 // Use the same cwd as the sidebar so both paths share GitManager's PR cache.104 // Use the same cwd as the sidebar so both paths share GitManager's PR cache.
55 const lookupCwdByThreadId = new Map<string, string>();105 const lookupCwdByThreadId = new Map<string, string>();
56 yield* Effect.forEach(106 yield* Effect.forEach(
57 candidates,107 lookupCandidates,
58 (thread) =>108 (thread) =>
59 Effect.gen(function* () {109 Effect.gen(function* () {
60 const project = projects.get(thread.projectId);110 const project = projects.get(thread.projectId);
...@@ -99,7 +149,7 @@ export const make = Effect.gen(function* () {...@@ -99,7 +149,7 @@ export const make = Effect.gen(function* () {
99 cwd === undefined ? ["missing-project", thread.id] : ["branch", cwd, thread.branch],149 cwd === undefined ? ["missing-project", thread.id] : ["branch", cwd, thread.branch],
100 );150 );
101 };151 };
102 const groups = Map.groupBy(candidates, lookupKey);152 const groups = Map.groupBy(lookupCandidates, lookupKey);
103153
104 const pullRequestFor = Effect.fn("ThreadSettlementReactor.pullRequestFor")(function* (154 const pullRequestFor = Effect.fn("ThreadSettlementReactor.pullRequestFor")(function* (
105 thread: (typeof candidates)[number],155 thread: (typeof candidates)[number],
...@@ -152,42 +202,9 @@ export const make = Effect.gen(function* () {...@@ -152,42 +202,9 @@ export const make = Effect.gen(function* () {
152 (group) =>202 (group) =>
153 Effect.gen(function* () {203 Effect.gen(function* () {
154 const pullRequest = yield* pullRequestFor(group[0]!);204 const pullRequest = yield* pullRequestFor(group[0]!);
155 yield* Effect.forEach(205 yield* Effect.forEach(group, (thread) => settleThread(thread, pullRequest), {
156 group,206 discard: true,
157 (thread) =>207 });
158 Effect.gen(function* () {
159 const settings = yield* settingsService.getSettings;
160 const decisionNow = DateTime.formatIso(yield* DateTime.now);
161 const settledAt = resolveAutoSettlementAt({
162 thread,
163 pullRequest,
164 now: decisionNow,
165 autoSettleAfterDays: settings.sidebarAutoSettleAfterDays,
166 autoSettleOnMerge: settings.sidebarAutoSettleOnMerge,
167 });
168 if (settledAt === null) {
169 return;
170 }
171 const uuid = yield* crypto.randomUUIDv4;
172 yield* engine.dispatch({
173 type: "thread.auto-settle",
174 commandId: CommandId.make(`server:auto-settle:${thread.id}:${uuid}`),
175 threadId: thread.id,
176 snapshotSequence: snapshot.snapshotSequence,
177 settledAt,
178 });
179 }).pipe(
180 Effect.catchCause((cause) =>
181 Cause.hasInterruptsOnly(cause)
182 ? Effect.failCause(cause)
183 : Effect.logWarning("automatic thread settlement skipped", {
184 threadId: thread.id,
185 cause: Cause.pretty(cause),
186 }),
187 ),
188 ),
189 { discard: true },
190 );
191 }).pipe(208 }).pipe(
192 Effect.catchCause((cause) =>209 Effect.catchCause((cause) =>
193 Cause.hasInterruptsOnly(cause)210 Cause.hasInterruptsOnly(cause)