authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-16 22:22:08-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-16 22:22:08-03:00
log19ee856ec099dddb94e28a3f933ec7c2575ed5a2
tree296369489e45cd8f3d7d1682f8036233d64216e0
parentc12ba7581ac00ba2a623ad23283da5d03d34c2d8
signature Signed by PGP key B5690EEEBB952194

fix(server): propagate linked pr changes and settle threads immediately (#12161)

8 files changed, 616 insertions(+), 101 deletions(-)

apps/server/src/orchestration/PullRequestSyncReactor.test.ts+155-9
...@@ -1,9 +1,11 @@...@@ -1,9 +1,11 @@
1import {1import {
2 EventId,
2 ProjectId,3 ProjectId,
3 ProviderInstanceId,4 ProviderInstanceId,
4 PullRequestOperationError,5 PullRequestOperationError,
5 ThreadId,6 ThreadId,
6 type OrchestrationCommand,7 type OrchestrationCommand,
8 type OrchestrationEvent,
7 type OrchestrationProjectShell,9 type OrchestrationProjectShell,
8 type OrchestrationShellSnapshot,10 type OrchestrationShellSnapshot,
9 type OrchestrationThreadShell,11 type OrchestrationThreadShell,
...@@ -18,6 +20,7 @@ import * as Crypto from "effect/Crypto";...@@ -18,6 +20,7 @@ import * as Crypto from "effect/Crypto";
18import * as Deferred from "effect/Deferred";20import * as Deferred from "effect/Deferred";
19import * as Effect from "effect/Effect";21import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";22import * as Layer from "effect/Layer";
23import * as PubSub from "effect/PubSub";
21import * as Queue from "effect/Queue";24import * as Queue from "effect/Queue";
22import * as Ref from "effect/Ref";25import * as Ref from "effect/Ref";
23import * as Stream from "effect/Stream";26import * as Stream from "effect/Stream";
...@@ -31,6 +34,7 @@ import {...@@ -31,6 +34,7 @@ import {
31} from "./Services/OrchestrationEngine.ts";34} from "./Services/OrchestrationEngine.ts";
32import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";35import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts";
33import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts";36import * as PullRequestSyncReactor from "./PullRequestSyncReactor.ts";
37import { resolveAutoSettlementAt } from "./ThreadSettlementPolicy.ts";
3438
35const NOW = "2026-08-28T12:00:00.000Z";39const NOW = "2026-08-28T12:00:00.000Z";
36const PROJECT_ID = ProjectId.make("sync-project");40const PROJECT_ID = ProjectId.make("sync-project");
...@@ -152,6 +156,7 @@ function makeSummary(...@@ -152,6 +156,7 @@ function makeSummary(
152}156}
153157
154interface HarnessOptions {158interface HarnessOptions {
159 readonly onDispatch?: (command: SyncCommand | LinkCommand) => Effect.Effect<void>;
155 readonly invalidate?: PullRequestService["Service"]["invalidate"];160 readonly invalidate?: PullRequestService["Service"]["invalidate"];
156 readonly snapshot: OrchestrationShellSnapshot;161 readonly snapshot: OrchestrationShellSnapshot;
157 readonly summary?: (162 readonly summary?: (
...@@ -165,6 +170,7 @@ interface HarnessOptions {...@@ -165,6 +170,7 @@ interface HarnessOptions {
165const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: HarnessOptions) {170const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: HarnessOptions) {
166 const activation = yield* Deferred.make<void>();171 const activation = yield* Deferred.make<void>();
167 const snapshots = yield* Ref.make(options.snapshot);172 const snapshots = yield* Ref.make(options.snapshot);
173 const events = yield* PubSub.unbounded<OrchestrationEvent>();
168 const snapshotReads = yield* Queue.unbounded<void>();174 const snapshotReads = yield* Queue.unbounded<void>();
169 const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);175 const syncCommands = yield* Ref.make<ReadonlyArray<SyncCommand>>([]);
170 const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);176 const linkCommands = yield* Ref.make<ReadonlyArray<LinkCommand>>([]);
...@@ -188,11 +194,13 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:...@@ -188,11 +194,13 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
188 const dispatch: OrchestrationEngineShape["dispatch"] = (command) => {194 const dispatch: OrchestrationEngineShape["dispatch"] = (command) => {
189 if (command.type === "thread.pull-request-link.sync") {195 if (command.type === "thread.pull-request-link.sync") {
190 return Ref.update(syncCommands, (recorded) => [...recorded, command]).pipe(196 return Ref.update(syncCommands, (recorded) => [...recorded, command]).pipe(
197 Effect.andThen(options.onDispatch?.(command) ?? Effect.void),
191 Effect.as({ sequence: 1 }),198 Effect.as({ sequence: 1 }),
192 );199 );
193 }200 }
194 if (command.type === "thread.pull-request.link") {201 if (command.type === "thread.pull-request.link") {
195 return Ref.update(linkCommands, (recorded) => [...recorded, command]).pipe(202 return Ref.update(linkCommands, (recorded) => [...recorded, command]).pipe(
203 Effect.andThen(options.onDispatch?.(command) ?? Effect.void),
196 Effect.as({ sequence: 1 }),204 Effect.as({ sequence: 1 }),
197 );205 );
198 }206 }
...@@ -213,6 +221,9 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:...@@ -213,6 +221,9 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
213 readEvents: () => Stream.empty,221 readEvents: () => Stream.empty,
214 dispatch,222 dispatch,
215 streamDomainEvents: Stream.empty,223 streamDomainEvents: Stream.empty,
224 subscribeDomainEvents: PubSub.subscribe(events).pipe(
225 Effect.map((subscription) => Stream.fromSubscription(subscription)),
226 ),
216 latestSequence: Effect.succeed(0),227 latestSequence: Effect.succeed(0),
217 }),228 }),
218 Layer.succeed(ServerActivation, Deferred.await(activation)),229 Layer.succeed(ServerActivation, Deferred.await(activation)),
...@@ -220,6 +231,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:...@@ -220,6 +231,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options:
220 );231 );
221232
222 return {233 return {
234 events,
223 activation,235 activation,
224 snapshots,236 snapshots,
225 snapshotReads,237 snapshotReads,
...@@ -274,6 +286,42 @@ function applySync(...@@ -274,6 +286,42 @@ function applySync(
274}286}
275287
276describe("PullRequestSyncReactor", () => {288describe("PullRequestSyncReactor", () => {
289 it.effect("syncs a newly linked merged PR without waiting for the periodic sweep", () =>
290 Effect.scoped(
291 Effect.gen(function* () {
292 yield* TestClock.setTime(Date.parse(NOW));
293 const fixture = yield* makeHarness({
294 snapshot: makeSnapshot([makeThread("one")]),
295 summary: (input) =>
296 Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
297 });
298 yield* Effect.gen(function* () {
299 const reactor = yield* startAndSweep(fixture);
300 const link = makeLink(42);
301 yield* Ref.set(
302 fixture.snapshots,
303 makeSnapshot([makeThread("one", { pullRequests: [link] })]),
304 );
305 yield* PubSub.publish(fixture.events, {
306 type: "thread.pull-request-linked",
307 sequence: 2,
308 eventId: EventId.make("linked"),
309 aggregateKind: "thread",
310 aggregateId: ThreadId.make("one"),
311 occurredAt: NOW,
312 commandId: null,
313 causationEventId: null,
314 correlationId: null,
315 metadata: {},
316 payload: { threadId: ThreadId.make("one"), link, updatedAt: NOW },
317 });
318 yield* Queue.take(fixture.snapshotReads);
319 yield* reactor.drain;
320 assert.strictEqual((yield* Ref.get(fixture.syncCommands))[0]?.snapshot.state, "merged");
321 }).pipe(Effect.provide(fixture.layer));
322 }),
323 ),
324 );
277 it.effect("retries a failed stack read after the summary becomes terminal", () =>325 it.effect("retries a failed stack read after the summary becomes terminal", () =>
278 Effect.scoped(326 Effect.scoped(
279 Effect.gen(function* () {327 Effect.gen(function* () {
...@@ -303,6 +351,7 @@ describe("PullRequestSyncReactor", () => {...@@ -303,6 +351,7 @@ describe("PullRequestSyncReactor", () => {
303 yield* Effect.gen(function* () {351 yield* Effect.gen(function* () {
304 const reactor = yield* startAndSweep(fixture);352 const reactor = yield* startAndSweep(fixture);
305 const commands = yield* Ref.get(fixture.syncCommands);353 const commands = yield* Ref.get(fixture.syncCommands);
354 assert.deepStrictEqual(commands, []);
306 yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands));355 yield* Ref.update(fixture.snapshots, (snapshot) => applySync(snapshot, commands));
307 yield* sweepAgain(fixture, reactor);356 yield* sweepAgain(fixture, reactor);
308 assert.strictEqual(attempts, 2);357 assert.strictEqual(attempts, 2);
...@@ -315,6 +364,85 @@ describe("PullRequestSyncReactor", () => {...@@ -315,6 +364,85 @@ describe("PullRequestSyncReactor", () => {
315 ),364 ),
316 );365 );
317366
367 it.effect("retries a failed sibling link before publishing a terminal snapshot", () =>
368 Effect.scoped(
369 Effect.gen(function* () {
370 yield* TestClock.setTime(Date.parse(NOW));
371 let failSibling = true;
372 const fixture = yield* makeHarness({
373 snapshot: makeSnapshot([
374 makeThread("one", { pullRequests: [makeLink(7, { state: "closed" })] }),
375 ]),
376 summary: (input) =>
377 Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
378 stack: () =>
379 Effect.succeed({
380 id: "stack",
381 number: 7,
382 url: "https://github.com/owner/repository/stacks/7",
383 base: "main",
384 layers: [
385 { number: 7, headBranch: "feature", state: "merged" },
386 { number: 8, headBranch: "sibling", state: "open" },
387 ],
388 }),
389 onDispatch: (command) =>
390 command.type === "thread.pull-request.link" && failSibling
391 ? Effect.die("temporary link failure")
392 : Effect.void,
393 });
394 yield* Effect.gen(function* () {
395 const reactor = yield* startAndSweep(fixture);
396 assert.deepStrictEqual(yield* Ref.get(fixture.syncCommands), []);
397 failSibling = false;
398 yield* sweepAgain(fixture, reactor);
399 assert.strictEqual((yield* Ref.get(fixture.syncCommands))[0]?.snapshot.state, "merged");
400 assert.strictEqual((yield* Ref.get(fixture.linkCommands)).at(-1)?.number, 8);
401 }).pipe(Effect.provide(fixture.layer));
402 }),
403 ),
404 );
405
406 it.effect("concurrent stack reads persist a shared sibling once before syncing both roots", () =>
407 Effect.scoped(
408 Effect.gen(function* () {
409 const readsReady = yield* Deferred.make<void>();
410 let reads = 0;
411 let linked = false;
412 const fixture = yield* makeHarness({
413 snapshot: makeSnapshot([makeThread("one", { pullRequests: [makeLink(7), makeLink(8)] })]),
414 stack: () =>
415 Effect.gen(function* () {
416 if (++reads === 2) yield* Deferred.succeed(readsReady, undefined);
417 yield* Deferred.await(readsReady);
418 return {
419 id: "stack",
420 number: 7,
421 url: "https://github.com/owner/repository/stacks/7",
422 base: "main",
423 layers: [{ number: 9, headBranch: "sibling", state: "open" as const }],
424 };
425 }),
426 onDispatch: (command) =>
427 Effect.gen(function* () {
428 if (command.type === "thread.pull-request.link") {
429 yield* Effect.yieldNow;
430 assert.strictEqual(linked, false);
431 linked = true;
432 } else {
433 assert.strictEqual(linked, true);
434 }
435 }),
436 });
437 yield* Effect.gen(function* () {
438 yield* startAndSweep(fixture);
439 assert.strictEqual((yield* Ref.get(fixture.linkCommands)).length, 1);
440 assert.strictEqual((yield* Ref.get(fixture.syncCommands)).length, 2);
441 }).pipe(Effect.provide(fixture.layer));
442 }),
443 ),
444 );
445
318 it.effect("explicit refresh reads a changed stack even when its PR summary is unchanged", () =>446 it.effect("explicit refresh reads a changed stack even when its PR summary is unchanged", () =>
319 Effect.scoped(447 Effect.scoped(
320 Effect.gen(function* () {448 Effect.gen(function* () {
...@@ -656,20 +784,38 @@ describe("PullRequestSyncReactor", () => {...@@ -656,20 +784,38 @@ describe("PullRequestSyncReactor", () => {
656 base: "main",784 base: "main",
657 layers: [785 layers: [
658 { number: 41, headBranch: "layer-1", state: "merged" },786 { number: 41, headBranch: "layer-1", state: "merged" },
659 { number: 42, headBranch: "layer-2", state: "open" },787 { number: 42, headBranch: "layer-2", state: "merged" },
660 { number: 43, headBranch: "layer-3", state: "open" },788 { number: 43, headBranch: "layer-3", state: "open" },
661 ],789 ],
662 };790 };
791 let thread = makeThread("one", {
792 pullRequests: [
793 makeLink(42, { state: "open" }),
794 makeLink(41, { state: "merged" }, { source: "stack-dismissed" }),
795 ],
796 });
663 const fixture = yield* makeHarness({797 const fixture = yield* makeHarness({
664 snapshot: makeSnapshot([798 snapshot: makeSnapshot([thread]),
665 makeThread("one", {799 summary: (input) =>
666 pullRequests: [800 Effect.succeed(makeSummary(input, { state: "merged", mergedAt: NOW })),
667 makeLink(42),
668 makeLink(41, { state: "merged" }, { source: "stack-dismissed" }),
669 ],
670 }),
671 ]),
672 stack: () => Effect.succeed(stack),801 stack: () => Effect.succeed(stack),
802 onDispatch: (command) =>
803 Effect.sync(() => {
804 thread =
805 command.type === "thread.pull-request.link"
806 ? { ...thread, pullRequests: [...thread.pullRequests, makeLink(command.number)] }
807 : applySync(makeSnapshot([thread]), [command]).threads[0]!;
808 // Every projected event may wake settlement, including the terminal root update.
809 assert.isNull(
810 resolveAutoSettlementAt({
811 thread,
812 pullRequest: null,
813 now: NOW,
814 autoSettleAfterDays: null,
815 autoSettleOnMerge: true,
816 }),
817 );
818 }),
673 });819 });
674820
675 yield* Effect.gen(function* () {821 yield* Effect.gen(function* () {
apps/server/src/orchestration/PullRequestSyncReactor.ts+51-43
...@@ -23,6 +23,8 @@ import * as Effect from "effect/Effect";...@@ -23,6 +23,8 @@ import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";23import * as Layer from "effect/Layer";
24import * as Schedule from "effect/Schedule";24import * as Schedule from "effect/Schedule";
25import type * as Scope from "effect/Scope";25import type * as Scope from "effect/Scope";
26import * as Semaphore from "effect/Semaphore";
27import * as Stream from "effect/Stream";
2628
27import * as PullRequestService from "../pullRequest/PullRequestService.ts";29import * as PullRequestService from "../pullRequest/PullRequestService.ts";
28import { forkParked } from "../serverActivation.ts";30import { forkParked } from "../serverActivation.ts";
...@@ -150,7 +152,7 @@ export const make = Effect.gen(function* () {...@@ -150,7 +152,7 @@ export const make = Effect.gen(function* () {
150 <E>(cause: Cause.Cause<E>): Effect.Effect<void, E> =>152 <E>(cause: Cause.Cause<E>): Effect.Effect<void, E> =>
151 Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);153 Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) : Effect.logWarning(message, fields);
152154
153 const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* () {155 const sweep = Effect.fn("PullRequestSyncReactor.sweep")(function* (requestedKey?: string) {
154 const snapshot = yield* snapshots.getShellSnapshot();156 const snapshot = yield* snapshots.getShellSnapshot();
155 const now = yield* DateTime.now;157 const now = yield* DateTime.now;
156 const nowMs = DateTime.toEpochMillis(now);158 const nowMs = DateTime.toEpochMillis(now);
...@@ -174,6 +176,7 @@ export const make = Effect.gen(function* () {...@@ -174,6 +176,7 @@ export const make = Effect.gen(function* () {
174 // Layers auto-linked this sweep, so two links of one thread that share a176 // Layers auto-linked this sweep, so two links of one thread that share a
175 // stack do not both try to add the same sibling.177 // stack do not both try to add the same sibling.
176 const linkedThisSweep = new Set<string>();178 const linkedThisSweep = new Set<string>();
179 const persistence = yield* Semaphore.make(1);
177180
178 const syncEntry = Effect.fn("PullRequestSyncReactor.syncEntry")(function* (181 const syncEntry = Effect.fn("PullRequestSyncReactor.syncEntry")(function* (
179 entry: LinkEntry,182 entry: LinkEntry,
...@@ -186,21 +189,8 @@ export const make = Effect.gen(function* () {...@@ -186,21 +189,8 @@ export const make = Effect.gen(function* () {
186 link.snapshot === null ||189 link.snapshot === null ||
187 !snapshotFieldsEqual(link.snapshot, fields) ||190 !snapshotFieldsEqual(link.snapshot, fields) ||
188 !stacksEqual(link.stack, nextStack);191 !stacksEqual(link.stack, nextStack);
189 if (changed) {192 // Persist discovered siblings before a terminal snapshot can trigger settlement.
190 const uuid = yield* crypto.randomUUIDv4;193 for (const layer of fetchedStack?.stack?.layers ?? []) {
191 yield* engine.dispatch({
192 type: "thread.pull-request-link.sync",
193 commandId: CommandId.make(`server:pr-sync:${thread.id}:${uuid}`),
194 threadId: thread.id,
195 host: normalizeThreadPullRequestKey(link).host,
196 repository: link.repository,
197 number: link.number,
198 snapshot: { ...fields, syncedAt: nowIso },
199 stack: nextStack,
200 });
201 }
202 if (fetchedStack === null || fetchedStack.stack === null) return;
203 for (const layer of fetchedStack.stack.layers) {
204 const layerKey = {194 const layerKey = {
205 host: normalizeThreadPullRequestKey(link).host,195 host: normalizeThreadPullRequestKey(link).host,
206 repository: link.repository,196 repository: link.repository,
...@@ -216,25 +206,29 @@ export const make = Effect.gen(function* () {...@@ -216,25 +206,29 @@ export const make = Effect.gen(function* () {
216 }206 }
217 const url = siblingPullRequestUrl(link.url, layer.number);207 const url = siblingPullRequestUrl(link.url, layer.number);
218 if (url === null) continue;208 if (url === null) continue;
209 const uuid = yield* crypto.randomUUIDv4;
210 yield* engine.dispatch({
211 type: "thread.pull-request.link",
212 commandId: CommandId.make(`server:pr-stack-link:${thread.id}:${uuid}`),
213 threadId: thread.id,
214 ...layerKey,
215 url,
216 source: "stack",
217 });
219 linkedThisSweep.add(dedupeKey);218 linkedThisSweep.add(dedupeKey);
219 }
220 if (changed) {
220 const uuid = yield* crypto.randomUUIDv4;221 const uuid = yield* crypto.randomUUIDv4;
221 yield* engine222 yield* engine.dispatch({
222 .dispatch({223 type: "thread.pull-request-link.sync",
223 type: "thread.pull-request.link",224 commandId: CommandId.make(`server:pr-sync:${thread.id}:${uuid}`),
224 commandId: CommandId.make(`server:pr-stack-link:${thread.id}:${uuid}`),225 threadId: thread.id,
225 threadId: thread.id,226 host: normalizeThreadPullRequestKey(link).host,
226 ...layerKey,227 repository: link.repository,
227 url,228 number: link.number,
228 source: "stack",229 snapshot: { ...fields, syncedAt: nowIso },
229 })230 stack: nextStack,
230 .pipe(231 });
231 Effect.catchCause(
232 logSkipped("pull request stack layer link skipped", {
233 threadId: thread.id,
234 number: layer.number,
235 }),
236 ),
237 );
238 }232 }
239 });233 });
240234
...@@ -275,8 +269,11 @@ export const make = Effect.gen(function* () {...@@ -275,8 +269,11 @@ export const make = Effect.gen(function* () {
275 )269 )
276 : null;270 : null;
277 if (needsStack) {271 if (needsStack) {
278 if (fetchedStack === null) retryStacks.add(key);272 if (fetchedStack === null) {
279 else retryStacks.delete(key);273 retryStacks.add(key);
274 return;
275 }
276 retryStacks.delete(key);
280 }277 }
281 // The host answered, so the cadence clock ticks even if a dispatch below is rejected.278 // The host answered, so the cadence clock ticks even if a dispatch below is rejected.
282 lastSyncedAt.set(key, nowMs);279 lastSyncedAt.set(key, nowMs);
...@@ -286,9 +283,13 @@ export const make = Effect.gen(function* () {...@@ -286,9 +283,13 @@ export const make = Effect.gen(function* () {
286 entries,283 entries,
287 (entry) =>284 (entry) =>
288 syncEntry(entry, fields, fetchedStack).pipe(285 syncEntry(entry, fields, fetchedStack).pipe(
289 Effect.catchCause(286 persistence.withPermits(1),
290 logSkipped("pull request sync skipped", { threadId: entry.thread.id, key }),287 Effect.catchCause((cause) => {
291 ),288 if (!Cause.hasInterruptsOnly(cause)) retryStacks.add(key);
289 return logSkipped("pull request sync skipped", { threadId: entry.thread.id, key })(
290 cause,
291 );
292 }),
292 ),293 ),
293 { discard: true },294 { discard: true },
294 );295 );
...@@ -297,7 +298,7 @@ export const make = Effect.gen(function* () {...@@ -297,7 +298,7 @@ export const make = Effect.gen(function* () {
297 yield* Effect.forEach(298 yield* Effect.forEach(
298 groups,299 groups,
299 ([key, entries]) =>300 ([key, entries]) =>
300 isDue(key, entries, nowMs)301 (requestedKey === undefined || requestedKey === key) && isDue(key, entries, nowMs)
301 ? syncGroup(key, entries).pipe(302 ? syncGroup(key, entries).pipe(
302 Effect.catchCause(logSkipped("pull request sync skipped", { key })),303 Effect.catchCause(logSkipped("pull request sync skipped", { key })),
303 )304 )
...@@ -306,13 +307,19 @@ export const make = Effect.gen(function* () {...@@ -306,13 +307,19 @@ export const make = Effect.gen(function* () {
306 );307 );
307 });308 });
308309
309 const worker = yield* makeDrainableWorker(() =>310 const worker = yield* makeDrainableWorker((key: string | undefined) =>
310 sweep().pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))),311 sweep(key).pipe(Effect.catchCause(logSkipped("pull request sync sweep failed", {}))),
311 );312 );
312313
313 const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn(314 const start: PullRequestSyncReactor["Service"]["start"] = Effect.fn(
314 "PullRequestSyncReactor.start",315 "PullRequestSyncReactor.start",
315 )(function* () {316 )(function* () {
317 const events = yield* engine.subscribeDomainEvents;
318 yield* forkParked(
319 Stream.runForEach(events, (event) =>
320 event.type === "thread.pull-request-linked" ? requestSync(event.payload.link) : Effect.void,
321 ),
322 );
316 yield* forkParked(323 yield* forkParked(
317 Effect.gen(function* () {324 Effect.gen(function* () {
318 yield* worker.enqueue(undefined);325 yield* worker.enqueue(undefined);
...@@ -323,8 +330,9 @@ export const make = Effect.gen(function* () {...@@ -323,8 +330,9 @@ export const make = Effect.gen(function* () {
323330
324 const requestSync: PullRequestSyncReactor["Service"]["requestSync"] = (key) =>331 const requestSync: PullRequestSyncReactor["Service"]["requestSync"] = (key) =>
325 Effect.suspend(() => {332 Effect.suspend(() => {
326 requested.set(threadPullRequestKeyOf(key), ++requestGeneration);333 const syncKey = threadPullRequestKeyOf(key);
327 return worker.enqueue(undefined);334 requested.set(syncKey, ++requestGeneration);
335 return worker.enqueue(syncKey);
328 });336 });
329337
330 return { start, drain: worker.drain, requestSync } satisfies PullRequestSyncReactor["Service"];338 return { start, drain: worker.drain, requestSync } satisfies PullRequestSyncReactor["Service"];
apps/server/src/orchestration/ThreadSettlementReactor.test.ts+88-8
...@@ -1,10 +1,12 @@...@@ -1,10 +1,12 @@
1import {1import {
2 DEFAULT_SERVER_SETTINGS,2 DEFAULT_SERVER_SETTINGS,
3 EventId,
3 ProjectId,4 ProjectId,
4 ProviderInstanceId,5 ProviderInstanceId,
5 PullRequestOperationError,6 PullRequestOperationError,
6 ThreadId,7 ThreadId,
7 type OrchestrationCommand,8 type OrchestrationCommand,
9 type OrchestrationEvent,
8 type OrchestrationProjectShell,10 type OrchestrationProjectShell,
9 type OrchestrationShellSnapshot,11 type OrchestrationShellSnapshot,
10 type OrchestrationThreadShell,12 type OrchestrationThreadShell,
...@@ -171,6 +173,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -171,6 +173,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
171 const settingsReads = yield* Queue.unbounded<ServerSettings>();173 const settingsReads = yield* Queue.unbounded<ServerSettings>();
172 const settingsChanges = yield* PubSub.unbounded<ServerSettings>();174 const settingsChanges = yield* PubSub.unbounded<ServerSettings>();
173 const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>();175 const mergedPullRequests = yield* PubSub.unbounded<PullRequestMergeEvent>();
176 const domainEvents = yield* PubSub.unbounded<OrchestrationEvent>();
174 const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]);177 const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]);
175 const branchCalls = yield* Ref.make<178 const branchCalls = yield* Ref.make<
176 ReadonlyArray<{ readonly cwd: string; readonly branch: string }>179 ReadonlyArray<{ readonly cwd: string; readonly branch: string }>
...@@ -258,6 +261,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -258,6 +261,9 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
258 readEvents: () => Stream.empty,261 readEvents: () => Stream.empty,
259 dispatch,262 dispatch,
260 streamDomainEvents: Stream.empty,263 streamDomainEvents: Stream.empty,
264 subscribeDomainEvents: PubSub.subscribe(domainEvents).pipe(
265 Effect.map((subscription) => Stream.fromSubscription(subscription)),
266 ),
261 latestSequence: Effect.succeed(0),267 latestSequence: Effect.succeed(0),
262 }),268 }),
263 Layer.succeed(ServerSettingsService, serverSettings),269 Layer.succeed(ServerSettingsService, serverSettings),
...@@ -280,6 +286,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:...@@ -280,6 +286,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
280 summaryRecovery,286 summaryRecovery,
281 invalidatedCwds,287 invalidatedCwds,
282 updateSettings,288 updateSettings,
289 publishEvent: (event: OrchestrationEvent) => PubSub.publish(domainEvents, event),
283 publishMerge: PubSub.publish(mergedPullRequests, {290 publishMerge: PubSub.publish(mergedPullRequests, {
284 projectId: PROJECT_ID,291 projectId: PROJECT_ID,
285 repository: "owner/repository",292 repository: "owner/repository",
...@@ -332,12 +339,12 @@ describe("ThreadSettlementReactor", () => {...@@ -332,12 +339,12 @@ describe("ThreadSettlementReactor", () => {
332 });339 });
333340
334 it.effect(341 it.effect(
335 "settles all-terminal links from snapshots and keeps open or unsynced links active",342 "settles synced terminal links immediately while open, unsynced, or running threads wait",
336 () =>343 () =>
337 Effect.scoped(344 Effect.scoped(
338 Effect.gen(function* () {345 Effect.gen(function* () {
339 yield* TestClock.setTime(Date.parse(NOW));346 yield* TestClock.setTime(Date.parse(NOW));
340 const link = (number: number, state: "open" | "merged" | null) => ({347 const link = (number: number, state: "open" | "closed" | "merged" | null) => ({
341 host: "example.test",348 host: "example.test",
342 repository: "owner/repository",349 repository: "owner/repository",
343 number,350 number,
...@@ -357,14 +364,30 @@ describe("ThreadSettlementReactor", () => {...@@ -357,14 +364,30 @@ describe("ThreadSettlementReactor", () => {
357 updatedAt: NOW,364 updatedAt: NOW,
358 syncedAt: NOW,365 syncedAt: NOW,
359 mergedAt: state === "merged" ? NOW : null,366 mergedAt: state === "merged" ? NOW : null,
367 closedAt: state === "closed" ? NOW : null,
360 },368 },
361 });369 });
370 const runningSession = {
371 threadId: ThreadId.make("running"),
372 status: "running" as const,
373 providerName: "Codex",
374 runtimeMode: "full-access" as const,
375 activeTurnId: null,
376 lastError: null,
377 updatedAt: NOW,
378 };
379 const threads = [
380 makeThread("merged", { pullRequests: [link(1, "open"), link(2, "merged")] }),
381 makeThread("closed", { pullRequests: [link(1, "open")] }),
382 makeThread("open", { pullRequests: [link(1, "open"), link(2, "open")] }),
383 makeThread("unsynced", { pullRequests: [link(1, "open"), link(2, null)] }),
384 makeThread("running", {
385 pullRequests: [link(1, "open")],
386 session: runningSession,
387 }),
388 ];
362 const fixture = yield* makeHarness({389 const fixture = yield* makeHarness({
363 snapshot: makeSnapshot([390 snapshot: makeSnapshot(threads),
364 makeThread("merged", { pullRequests: [link(1, "merged"), link(2, "merged")] }),
365 makeThread("open", { pullRequests: [link(1, "merged"), link(2, "open")] }),
366 makeThread("unsynced", { pullRequests: [link(1, "merged"), link(2, null)] }),
367 ]),
368 settings: { ...DEFAULT_SERVER_SETTINGS, sidebarAutoSettleOnMerge: true },391 settings: { ...DEFAULT_SERVER_SETTINGS, sidebarAutoSettleOnMerge: true },
369 branchPullRequest: () => Effect.die("linked threads must not query the branch"),392 branchPullRequest: () => Effect.die("linked threads must not query the branch"),
370 pullRequestSummary: () => Effect.die("linked threads must use their snapshots"),393 pullRequestSummary: () => Effect.die("linked threads must use their snapshots"),
...@@ -372,9 +395,66 @@ describe("ThreadSettlementReactor", () => {...@@ -372,9 +395,66 @@ describe("ThreadSettlementReactor", () => {
372 yield* Effect.gen(function* () {395 yield* Effect.gen(function* () {
373 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;396 const reactor = yield* ThreadSettlementReactor.ThreadSettlementReactor;
374 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);397 yield* startHarness(reactor, fixture.activation, fixture.snapshotReads);
398 assert.deepStrictEqual(yield* Ref.get(fixture.commands), []);
399 const eventBase = {
400 sequence: 2,
401 eventId: EventId.make("pull-request-synced"),
402 aggregateKind: "thread" as const,
403 occurredAt: NOW,
404 commandId: null,
405 causationEventId: null,
406 correlationId: null,
407 metadata: {},
408 };
409 for (const thread of threads) {
410 const terminalLink = link(1, thread.id === "closed" ? "closed" : "merged");
411 yield* Ref.update(fixture.snapshots, (snapshot) => ({
412 ...snapshot,
413 threads: snapshot.threads.map((current) =>
414 current.id === thread.id
415 ? { ...current, pullRequests: [terminalLink, ...current.pullRequests.slice(1)] }
416 : current,
417 ),
418 }));
419 yield* fixture.publishEvent({
420 ...eventBase,
421 type: "thread.pull-request-synced",
422 aggregateId: thread.id,
423 payload: {
424 threadId: thread.id,
425 host: terminalLink.host,
426 repository: terminalLink.repository,
427 number: terminalLink.number,
428 snapshot: terminalLink.snapshot!,
429 stack: null,
430 updatedAt: NOW,
431 },
432 });
433 yield* Queue.take(fixture.snapshotReads);
434 yield* reactor.drain;
435 }
436 assert.deepStrictEqual(
437 (yield* Ref.get(fixture.commands)).map(({ threadId }) => threadId),
438 [ThreadId.make("merged"), ThreadId.make("closed")],
439 );
440 const readySession = { ...runningSession, status: "ready" as const };
441 yield* Ref.update(fixture.snapshots, (snapshot) => ({
442 ...snapshot,
443 threads: snapshot.threads.map((thread) =>
444 thread.id === readySession.threadId ? { ...thread, session: readySession } : thread,
445 ),
446 }));
447 yield* fixture.publishEvent({
448 ...eventBase,
449 type: "thread.session-set",
450 aggregateId: readySession.threadId,
451 payload: { threadId: readySession.threadId, session: readySession },
452 });
453 yield* Queue.take(fixture.snapshotReads);
454 yield* reactor.drain;
375 assert.deepStrictEqual(455 assert.deepStrictEqual(
376 (yield* Ref.get(fixture.commands)).map(({ threadId }) => threadId),456 (yield* Ref.get(fixture.commands)).map(({ threadId }) => threadId),
377 [ThreadId.make("merged")],457 [ThreadId.make("merged"), ThreadId.make("closed"), ThreadId.make("running")],
378 );458 );
379 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), []);459 assert.deepStrictEqual(yield* Ref.get(fixture.branchCalls), []);
380 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []);460 assert.deepStrictEqual(yield* Ref.get(fixture.summaryCalls), []);
apps/server/src/orchestration/ThreadSettlementReactor.ts+43-6
...@@ -1,4 +1,9 @@...@@ -1,4 +1,9 @@
1import { CommandId, type ServerSettings as ServerSettingsValue } from "@t3tools/contracts";1import {
2 CommandId,
3 type OrchestrationEvent,
4 type ServerSettings as ServerSettingsValue,
5 type ThreadId,
6} from "@t3tools/contracts";
2import { resolveProjectSettings } from "@t3tools/shared/projectSettings";7import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
3import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";8import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
4import * as Cause from "effect/Cause";9import * as Cause from "effect/Cause";
...@@ -84,6 +89,7 @@ export const make = Effect.gen(function* () {...@@ -84,6 +89,7 @@ export const make = Effect.gen(function* () {
8489
85 const sweep = Effect.fn("ThreadSettlementReactor.sweep")(function* (90 const sweep = Effect.fn("ThreadSettlementReactor.sweep")(function* (
86 mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,91 mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
92 threadId?: ThreadId,
87 ) {93 ) {
88 const settings = yield* settingsService.getSettings;94 const settings = yield* settingsService.getSettings;
89 if (!autoSettlementConfigured(settings)) {95 if (!autoSettlementConfigured(settings)) {
...@@ -94,7 +100,11 @@ export const make = Effect.gen(function* () {...@@ -94,7 +100,11 @@ export const make = Effect.gen(function* () {
94 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));100 const projects = new Map(snapshot.projects.map((project) => [project.id, project]));
95 // A merge rechecks all candidates, including branches that discovery has101 // A merge rechecks all candidates, including branches that discovery has
96 // not linked yet. Those lookups can still have cached the PR as open.102 // not linked yet. Those lookups can still have cached the PR as open.
97 const candidates = snapshot.threads.filter((thread) => isAutoSettlementCandidate(thread, now));103 const candidates = snapshot.threads.filter(
104 (thread) =>
105 (threadId === undefined || thread.id === threadId) &&
106 isAutoSettlementCandidate(thread, now),
107 );
98108
99 // Return the thread when it still needs a pull request decision. A rejected109 // Return the thread when it still needs a pull request decision. A rejected
100 // dispatch skips it for this snapshot instead of retrying through a lookup.110 // dispatch skips it for this snapshot instead of retrying through a lookup.
...@@ -279,8 +289,11 @@ export const make = Effect.gen(function* () {...@@ -279,8 +289,11 @@ export const make = Effect.gen(function* () {
279 );289 );
280 });290 });
281291
282 const runSweep = (mergedPullRequest: PullRequestService.PullRequestMergeEvent | null) =>292 const runSweep = (
283 sweep(mergedPullRequest).pipe(293 mergedPullRequest: PullRequestService.PullRequestMergeEvent | null,
294 threadId?: ThreadId,
295 ) =>
296 sweep(mergedPullRequest, threadId).pipe(
284 Effect.catchCause((cause) =>297 Effect.catchCause((cause) =>
285 Cause.hasInterruptsOnly(cause)298 Cause.hasInterruptsOnly(cause)
286 ? Effect.failCause(cause)299 ? Effect.failCause(cause)
...@@ -289,13 +302,36 @@ export const make = Effect.gen(function* () {...@@ -289,13 +302,36 @@ export const make = Effect.gen(function* () {
289 }),302 }),
290 ),303 ),
291 );304 );
292 const worker = yield* makeDrainableWorker(() => runSweep(null));305 const worker = yield* makeDrainableWorker((threadId: ThreadId | undefined) =>
306 runSweep(null, threadId),
307 );
308
309 const processEvent = (event: OrchestrationEvent) => {
310 switch (event.type) {
311 case "thread.pull-request-linked":
312 case "thread.pull-request-synced":
313 case "thread.pull-request-unlinked":
314 // Merge notifications can arrive before the linked snapshot is projected.
315 // Recheck the persisted state so terminal links settle without the timer.
316 return worker.enqueue(event.payload.threadId);
317 case "thread.session-set":
318 if (
319 event.payload.session.status !== "running" &&
320 event.payload.session.status !== "starting"
321 ) {
322 return worker.enqueue(event.payload.threadId);
323 }
324 break;
325 }
326 return Effect.void;
327 };
293328
294 const start: ThreadSettlementReactor["Service"]["start"] = Effect.fn(329 const start: ThreadSettlementReactor["Service"]["start"] = Effect.fn(
295 "ThreadSettlementReactor.start",330 "ThreadSettlementReactor.start",
296 )(function* () {331 )(function* () {
297 const settingsChanges = yield* settingsService.subscribeChanges;332 const settingsChanges = yield* settingsService.subscribeChanges;
298 const mergedPullRequests = yield* pullRequests.subscribeMerges;333 const mergedPullRequests = yield* pullRequests.subscribeMerges;
334 const events = yield* engine.subscribeDomainEvents;
299 const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);335 const initialSettings = yield* settingsService.getSettings.pipe(Effect.orDie);
300 let lastSettlementSettings = autoSettlementSettingsKey(initialSettings);336 let lastSettlementSettings = autoSettlementSettingsKey(initialSettings);
301 yield* forkParked(337 yield* forkParked(
...@@ -314,7 +350,8 @@ export const make = Effect.gen(function* () {...@@ -314,7 +350,8 @@ export const make = Effect.gen(function* () {
314 return worker.enqueue(undefined);350 return worker.enqueue(undefined);
315 }),351 }),
316 );352 );
317 yield* forkParked(Stream.runForEach(mergedPullRequests, runSweep));353 yield* forkParked(Stream.runForEach(mergedPullRequests, (event) => runSweep(event)));
354 yield* forkParked(Stream.runForEach(events, processEvent));
318 });355 });
319356
320 return { start, drain: worker.drain } satisfies ThreadSettlementReactor["Service"];357 return { start, drain: worker.drain } satisfies ThreadSettlementReactor["Service"];
apps/server/src/pullRequest/PullRequestService.test.ts+157-24
...@@ -1093,8 +1093,13 @@ it.effect("publishes a merge for immediate settlement only after host confirmati...@@ -1093,8 +1093,13 @@ it.effect("publishes a merge for immediate settlement only after host confirmati
10931093
1094 // Queueing succeeds while the host still reports an open PR.1094 // Queueing succeeds while the host still reports an open PR.
1095 yield* service.runAction({ ...reference, action: "merge" });1095 yield* service.runAction({ ...reference, action: "merge" });
1096 const queuedRefresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes));
1096 confirmationFails = true;1097 confirmationFails = true;
1097 yield* service.runAction({ ...reference, action: "merge" });1098 yield* service.runAction({ ...reference, action: "merge" });
1099 assert.isAbove(
1100 Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)),
1101 queuedRefresh,
1102 );
1098 confirmationFails = false;1103 confirmationFails = false;
1099 state = "merged";1104 state = "merged";
1100 yield* TestClock.setTime(Date.parse(mergedAt));1105 yield* TestClock.setTime(Date.parse(mergedAt));
...@@ -1113,6 +1118,53 @@ it.effect("publishes a merge for immediate settlement only after host confirmati...@@ -1113,6 +1118,53 @@ it.effect("publishes a merge for immediate settlement only after host confirmati
1113 ),1118 ),
1114);1119);
11151120
1121it.effect("refreshes every reader before a queued merge confirmation finishes", () =>
1122 Effect.scoped(
1123 Effect.gen(function* () {
1124 const confirmationStarted = yield* Deferred.make<void>();
1125 const confirm = yield* Deferred.make<void>();
1126 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
1127 const service = yield* makeService({
1128 projects: [
1129 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
1130 ],
1131 providers: [
1132 fakeProvider("github", {
1133 getChangeRequestSummary: () =>
1134 Effect.gen(function* () {
1135 yield* Deferred.succeed(confirmationStarted, undefined);
1136 yield* Deferred.await(confirm);
1137 return changeRequest(1, "2026-09-16T00:00:00.000Z");
1138 }),
1139 }),
1140 ],
1141 });
1142 const merges = yield* service.subscribeMerges;
1143 const observedMerge = yield* Stream.runHead(merges).pipe(
1144 Effect.forkChild({ startImmediately: true }),
1145 );
1146 const readers = yield* Effect.forEach([0, 1], () =>
1147 Stream.runHead(service.subscribeRefreshes).pipe(
1148 Effect.forkChild({ startImmediately: true }),
1149 ),
1150 );
1151 const action = yield* service
1152 .runAction({ ...reference, action: "merge" })
1153 .pipe(Effect.forkChild({ startImmediately: true }));
1154 yield* Deferred.await(confirmationStarted);
1155 const revisions = yield* Effect.forEach(readers, (reader) =>
1156 Fiber.join(reader).pipe(Effect.map(Option.getOrThrow)),
1157 );
1158 assert.isAbove(revisions[0]!, 0);
1159 assert.strictEqual(revisions[0], revisions[1]);
1160 assert.isUndefined(action.pollUnsafe());
1161 yield* Deferred.succeed(confirm, undefined);
1162 yield* Fiber.join(action);
1163 assert.isUndefined(observedMerge.pollUnsafe());
1164 }),
1165 ),
1166);
1167
1116it.effect("refuses an action this viewer may not take, and says what access it takes", () =>1168it.effect("refuses an action this viewer may not take, and says what access it takes", () =>
1117 Effect.gen(function* () {1169 Effect.gen(function* () {
1118 let ran: string | null = null;1170 let ran: string | null = null;
...@@ -2378,10 +2430,16 @@ it.effect("invalidates the cached activity after reacting, like the other mutati...@@ -2378,10 +2430,16 @@ it.effect("invalidates the cached activity after reacting, like the other mutati
2378 ],2430 ],
2379 });2431 });
23802432
2433 yield* service.refreshAfterTurn(reference.projectId);
2434 const previousRefresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes));
2381 yield* service.activity(reference);2435 yield* service.activity(reference);
2382 assert.strictEqual(activityCalls, 1);2436 assert.strictEqual(activityCalls, 1);
23832437
2384 yield* service.setReaction({ ...reference, content: "heart", reacted: true });2438 yield* service.setReaction({ ...reference, content: "heart", reacted: true });
2439 assert.isAbove(
2440 Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)),
2441 previousRefresh,
2442 );
2385 yield* service.activity(reference);2443 yield* service.activity(reference);
23862444
2387 assert.strictEqual(activityCalls, 2);2445 assert.strictEqual(activityCalls, 2);
...@@ -3255,31 +3313,106 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag...@@ -3255,31 +3313,106 @@ it.effect("explicit and turn invalidations make the next listing ask the host ag
3255 }),3313 }),
3256);3314);
32573315
3258it.effect("a mutation makes the next listing ask the host again, with no client asking", () =>3316it.effect("close and reopen notify subscribed readers after invalidating their cached state", () =>
3259 Effect.gen(function* () {3317 Effect.scoped(
3260 let hostCalls = 0;3318 Effect.gen(function* () {
3261 const service = yield* makeService({3319 let hostCalls = 0;
3262 projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],3320 let state: "open" | "closed" = "open";
3263 providers: [3321 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
3264 fakeProvider("github", {3322 const service = yield* makeService({
3265 listChangeRequests: () => {3323 projects: [
3266 hostCalls += 1;3324 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
3267 return Effect.succeed({ items: [], truncated: false, continues: false });3325 ],
3268 },3326 providers: [
3269 }),3327 fakeProvider("github", {
3270 ],3328 getChangeRequestSummary: () =>
3271 });3329 Effect.succeed({ ...changeRequest(1, "2026-09-16T00:00:00.000Z"), state }),
3330 runAction: (input) =>
3331 Effect.sync(() => {
3332 state = input.action === "close" ? "closed" : "open";
3333 }),
3334 listChangeRequests: () => {
3335 hostCalls += 1;
3336 return Effect.succeed({ items: [], truncated: false, continues: false });
3337 },
3338 }),
3339 ],
3340 });
32723341
3273 yield* service.list({ state: "open" });3342 yield* service.refreshAfterTurn(reference.projectId);
3274 yield* service.runAction({3343 yield* service.list({ state: "open" });
3275 projectId: "p1" as ProjectId,3344 assert.strictEqual((yield* service.summary(reference)).state, "open");
3276 repository: "acme/web",3345 for (const action of ["close", "reopen"] as const) {
3277 number: 1,3346 const refreshed = yield* service.subscribeRefreshes.pipe(
3278 action: "close",3347 Stream.drop(1),
3279 });3348 Stream.take(1),
3280 yield* service.list({ state: "open" });3349 Stream.mapEffect(() =>
3281 assert.strictEqual(hostCalls, 2);3350 Effect.gen(function* () {
3282 }),3351 yield* service.list({ state: "open" });
3352 return yield* service.summary(reference);
3353 }),
3354 ),
3355 Stream.runHead,
3356 Effect.forkChild({ startImmediately: true }),
3357 );
3358 yield* service.runAction({ ...reference, action });
3359 assert.strictEqual(
3360 Option.getOrThrow(yield* Fiber.join(refreshed)).state,
3361 action === "close" ? "closed" : "open",
3362 );
3363 }
3364 assert.strictEqual(hostCalls, 3);
3365 }),
3366 ),
3367);
3368
3369it.effect("explicit invalidation refreshes origin readers after a routed host mutation", () =>
3370 Effect.scoped(
3371 Effect.gen(function* () {
3372 let state: "open" | "closed" = "open";
3373 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
3374 const service = yield* makeService({
3375 projects: [
3376 project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" }),
3377 ],
3378 providers: [
3379 fakeProvider("github", {
3380 getChangeRequestSummary: () =>
3381 Effect.succeed({ ...changeRequest(1, "2026-09-16T00:00:00.000Z"), state }),
3382 }),
3383 ],
3384 });
3385 yield* service.refreshAfterTurn(reference.projectId);
3386 let revision = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes));
3387 // The sync reactor invalidates before reading; it must not notify itself again.
3388 yield* service.invalidate({ reference });
3389 assert.strictEqual(
3390 Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)),
3391 revision,
3392 );
3393 assert.strictEqual((yield* service.summary(reference)).state, "open");
3394
3395 for (const nextState of ["closed", "open"] as const) {
3396 const refreshed = yield* service.subscribeRefreshes.pipe(
3397 Stream.drop(1),
3398 Stream.take(1),
3399 Stream.mapEffect((nextRevision) =>
3400 service
3401 .summary(reference)
3402 .pipe(Effect.map((summary) => ({ revision: nextRevision, state: summary.state }))),
3403 ),
3404 Stream.runHead,
3405 Effect.forkChild({ startImmediately: true }),
3406 );
3407 state = nextState;
3408 yield* service.invalidate({ reference }, { notifyReaders: true });
3409 const result = Option.getOrThrow(yield* Fiber.join(refreshed));
3410 assert.strictEqual(result.state, nextState);
3411 assert.isAbove(result.revision, revision);
3412 revision = result.revision;
3413 }
3414 }),
3415 ),
3283);3416);
32843417
3285it.effect("does not cache a failed listing", () =>3418it.effect("does not cache a failed listing", () =>
apps/server/src/pullRequest/PullRequestService.ts+18-10
...@@ -232,7 +232,10 @@ export class PullRequestService extends Context.Service<...@@ -232,7 +232,10 @@ export class PullRequestService extends Context.Service<
232 readonly setLabels: (232 readonly setLabels: (
233 input: PullRequestLabelChangeInput,233 input: PullRequestLabelChangeInput,
234 ) => Effect.Effect<void, PullRequestError>;234 ) => Effect.Effect<void, PullRequestError>;
235 readonly invalidate: (input: PullRequestInvalidateInput) => Effect.Effect<void>;235 readonly invalidate: (
236 input: PullRequestInvalidateInput,
237 options?: { readonly notifyReaders?: boolean },
238 ) => Effect.Effect<void>;
236 }239 }
237>()("t3/pullRequest/PullRequestService") {}240>()("t3/pullRequest/PullRequestService") {}
238241
...@@ -2839,10 +2842,12 @@ export const make = Effect.gen(function* () {...@@ -2839,10 +2842,12 @@ export const make = Effect.gen(function* () {
2839 return { stats: [...held, ...result.stats] };2842 return { stats: [...held, ...result.stats] };
2840 });2843 });
28412844
2842 const invalidate: PullRequestService["Service"]["invalidate"] = (input) => {2845 const invalidate: PullRequestService["Service"]["invalidate"] = Effect.fn(
2846 "PullRequestService.invalidate",
2847 )(function* (input, options) {
2843 const reference = input.reference;2848 const reference = input.reference;
2844 if (reference !== undefined) {2849 if (reference !== undefined) {
2845 return canonicalRef(reference).pipe(2850 yield* canonicalRef(reference).pipe(
2846 Effect.flatMap((ref) =>2851 Effect.flatMap((ref) =>
2847 readCache2852 readCache
2848 .invalidate(refScope(ref))2853 .invalidate(refScope(ref))
...@@ -2850,12 +2855,15 @@ export const make = Effect.gen(function* () {...@@ -2850,12 +2855,15 @@ export const make = Effect.gen(function* () {
2850 ),2855 ),
2851 Effect.ignore,2856 Effect.ignore,
2852 );2857 );
2853 }2858 } else {
2854 return Effect.sync(() => {
2855 listingsEpoch = ++epochCounter;2859 listingsEpoch = ++epochCounter;
2856 viewersByHost.clear();2860 viewersByHost.clear();
2857 }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights)));2861 yield* Cache.invalidateAll(viewerFlights);
2858 };2862 }
2863 if (options?.notifyReaders) {
2864 yield* SubscriptionRef.set(pullRequestRefreshes, ++epochCounter);
2865 }
2866 });
28592867
2860 const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = (projectId) =>2868 const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = (projectId) =>
2861 Effect.suspend(() => {2869 Effect.suspend(() => {
...@@ -2874,9 +2882,7 @@ export const make = Effect.gen(function* () {...@@ -2874,9 +2882,7 @@ export const make = Effect.gen(function* () {
2874 .pipe(Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, listingsEpoch)));2882 .pipe(Effect.andThen(SubscriptionRef.set(pullRequestRefreshes, listingsEpoch)));
2875 });2883 });
28762884
2877 // A mutation's own client re-reads right after it, and every other client's next read must2885 // Invalidate before notifying every client so mounted readers immediately fetch the edit.
2878 // see the action too — so a write forgets the change request it touched and the listings its
2879 // state change reorders, for everyone, without any client asking.
2880 const invalidatedByMutation =2886 const invalidatedByMutation =
2881 <I extends PullRequestRef>(2887 <I extends PullRequestRef>(
2882 method: (input: I) => Effect.Effect<void, PullRequestError>,2888 method: (input: I) => Effect.Effect<void, PullRequestError>,
...@@ -2894,6 +2900,7 @@ export const make = Effect.gen(function* () {...@@ -2894,6 +2900,7 @@ export const make = Effect.gen(function* () {
2894 }),2900 }),
2895 ),2901 ),
2896 );2902 );
2903 yield* SubscriptionRef.set(pullRequestRefreshes, listingsEpoch);
2897 });2904 });
2898 const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn(2905 const runActionAndInvalidate: PullRequestService["Service"]["runAction"] = Effect.fn(
2899 "PullRequestService.runActionAndInvalidate",2906 "PullRequestService.runActionAndInvalidate",
...@@ -2905,6 +2912,7 @@ export const make = Effect.gen(function* () {...@@ -2905,6 +2912,7 @@ export const make = Effect.gen(function* () {
2905 );2912 );
2906 bumpRefEpoch({ ...ref, repository });2913 bumpRefEpoch({ ...ref, repository });
2907 listingsEpoch = ++epochCounter;2914 listingsEpoch = ++epochCounter;
2915 yield* SubscriptionRef.set(pullRequestRefreshes, listingsEpoch);
2908 if (input.action === "merge") {2916 if (input.action === "merge") {
2909 // A successful merge action can merely enqueue the PR or enable auto-merge.2917 // A successful merge action can merely enqueue the PR or enable auto-merge.
2910 const confirmed = yield* summaryUncached({ ...input, repository }).pipe(2918 const confirmed = yield* summaryUncached({ ...input, repository }).pipe(
apps/server/src/ws.ts+1-1
...@@ -2808,7 +2808,7 @@ const makeWsRpcLayer = (...@@ -2808,7 +2808,7 @@ const makeWsRpcLayer = (
2808 [WS_METHODS.pullRequestsInvalidate]: (input) =>2808 [WS_METHODS.pullRequestsInvalidate]: (input) =>
2809 observeRpcEffect(2809 observeRpcEffect(
2810 WS_METHODS.pullRequestsInvalidate,2810 WS_METHODS.pullRequestsInvalidate,
2811 pullRequests.invalidate(input).pipe(2811 pullRequests.invalidate(input, { notifyReaders: true }).pipe(
2812 // A reader asking for fresh host state also wants the thread badges it feeds to2812 // A reader asking for fresh host state also wants the thread badges it feeds to
2813 // catch up, including a merged link the sweep would otherwise never revisit.2813 // catch up, including a merged link the sweep would otherwise never revisit.
2814 Effect.andThen(2814 Effect.andThen(
packages/client-runtime/src/state/pullRequests.test.ts+103
...@@ -34,6 +34,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts";...@@ -34,6 +34,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts";
34import type { WsRpcProtocolClient } from "../rpc/protocol.ts";34import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
35import type { RpcSession } from "../rpc/session.ts";35import type { RpcSession } from "../rpc/session.ts";
36import {36import {
37 createLinkedPullRequestSummaryAtomFamily,
37 createPullRequestEnvironmentAtoms,38 createPullRequestEnvironmentAtoms,
38 createPullRequestStackAtomFamily,39 createPullRequestStackAtomFamily,
39} from "./pullRequests.ts";40} from "./pullRequests.ts";
...@@ -808,6 +809,108 @@ it.effect("keeps concurrent diff file reads on different hosts separate", () =>...@@ -808,6 +809,108 @@ it.effect("keeps concurrent diff file reads on different hosts separate", () =>
808 ),809 ),
809);810);
810811
812it.effect("shares close, reopen, and merge with an untouched client's mounted PR readers", () =>
813 Effect.scoped(
814 Effect.gen(function* () {
815 const revision = yield* SubscriptionRef.make(0);
816 let state: "open" | "closed" | "merged" = "open";
817 const client = {
818 [WS_METHODS.pullRequestsSubscribeRefreshes]: () => SubscriptionRef.changes(revision),
819 [WS_METHODS.pullRequestsSummary]: () => Effect.sync(() => ({ state })),
820 [WS_METHODS.pullRequestsDetail]: () => Effect.sync(() => ({ state })),
821 [WS_METHODS.pullRequestsList]: () =>
822 Effect.sync(() => ({ entries: [{ number: 1, state }] })),
823 [WS_METHODS.pullRequestsRunAction]: (input: {
824 readonly action: "close" | "reopen" | "merge";
825 }) =>
826 Effect.gen(function* () {
827 state =
828 input.action === "close" ? "closed" : input.action === "merge" ? "merged" : "open";
829 yield* SubscriptionRef.update(revision, (value) => value + 1);
830 }),
831 } as unknown as WsRpcProtocolClient;
832 const writer = yield* makeTestRuntime(client);
833 const reader = yield* makeTestRuntime(client);
834 const target = {
835 environmentId: TARGET.environmentId,
836 input: {
837 projectId: ProjectId.make("project-1"),
838 host: "github.example.com",
839 repository: "acme/web",
840 number: 1,
841 },
842 };
843 const detail = reader.atoms.detail(target);
844 const summary = createLinkedPullRequestSummaryAtomFamily(
845 reader.runtime,
846 reader.atoms.refreshes,
847 )(target);
848 const list = reader.atoms.list({
849 environmentId: TARGET.environmentId,
850 input: { state: "all" },
851 });
852 const unmountDetail = reader.registry.mount(detail);
853 const unmountSummary = reader.registry.mount(summary);
854 const unmountList = reader.registry.mount(list);
855 yield* Effect.addFinalizer(() =>
856 Effect.sync(() => {
857 unmountDetail();
858 unmountSummary();
859 unmountList();
860 }),
861 );
862 expect((yield* AtomRegistry.getResult(reader.registry, detail)).state).toBe("open");
863 expect((yield* AtomRegistry.getResult(reader.registry, summary)).state).toBe("open");
864 expect((yield* AtomRegistry.getResult(reader.registry, list)).entries[0]?.state).toBe("open");
865
866 for (const [action, expected] of [
867 ["close", "closed"],
868 ["reopen", "open"],
869 ["merge", "merged"],
870 ] as const) {
871 const detailChanged = Latch.makeUnsafe();
872 const summaryChanged = Latch.makeUnsafe();
873 const listChanged = Latch.makeUnsafe();
874 const stops = [
875 reader.registry.subscribe(detail, (result) => {
876 if (AsyncResult.isSuccess(result) && result.value.state === expected) {
877 detailChanged.openUnsafe();
878 }
879 }),
880 reader.registry.subscribe(summary, (result) => {
881 if (AsyncResult.isSuccess(result) && result.value.state === expected) {
882 summaryChanged.openUnsafe();
883 }
884 }),
885 reader.registry.subscribe(list, (result) => {
886 if (AsyncResult.isSuccess(result) && result.value.entries[0]?.state === expected) {
887 listChanged.openUnsafe();
888 }
889 }),
890 ];
891 yield* Effect.addFinalizer(() => Effect.sync(() => stops.forEach((stop) => stop())));
892 const result = yield* Effect.promise(() =>
893 writer.atoms.runAction.run(writer.registry, {
894 ...target,
895 input: { ...target.input, action },
896 }),
897 );
898 expect(AsyncResult.isSuccess(result)).toBe(true);
899 // The second client receives only the server push: no local refresh or timer tick.
900 yield* detailChanged.await;
901 yield* summaryChanged.await;
902 yield* listChanged.await;
903 expect((yield* AtomRegistry.getResult(reader.registry, detail)).state).toBe(expected);
904 expect((yield* AtomRegistry.getResult(reader.registry, summary)).state).toBe(expected);
905 expect((yield* AtomRegistry.getResult(reader.registry, list)).entries[0]?.state).toBe(
906 expected,
907 );
908 stops.forEach((stop) => stop());
909 }
910 }),
911 ),
912);
913
811it.effect("refreshes pull request activity after a comment is updated", () =>914it.effect("refreshes pull request activity after a comment is updated", () =>
812 Effect.scoped(915 Effect.scoped(
813 Effect.gen(function* () {916 Effect.gen(function* () {