diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts index 8cc5808ed369253392a7243b62c90842e98f49d9..55c1866ee9991b9862c80f5ac3901e24815a4006 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.test.ts @@ -15,6 +15,7 @@ import { type PullRequestRef, type PullRequestStack, type PullRequestSummary, + type ThreadPullRequestKey, type ThreadPullRequestLink, type ThreadPullRequestSnapshot, } from "@t3tools/contracts"; @@ -181,6 +182,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: const summaryCalls = yield* Ref.make>([]); const stackCalls = yield* Ref.make>([]); const domainEvents = yield* Queue.unbounded(); + const stateChanges = yield* Queue.unbounded(); const summary: PullRequestService.PullRequestService["Service"]["summary"] = ( input, @@ -218,6 +220,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: summary, stack, invalidate: options.invalidate ?? (() => Effect.void), + subscribeStateChanges: Effect.succeed(Stream.fromQueue(stateChanges)), }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ // Mirrors the store's filter: active threads that have at least one link. @@ -269,6 +272,7 @@ const makeHarness = Effect.fn("makePullRequestSyncHarness")(function* (options: summaryCalls, stackCalls, domainEvents, + stateChanges, layer: PullRequestSyncReactor.layer.pipe(Layer.provide(layerDependencies)), }; }); @@ -654,6 +658,49 @@ describe("PullRequestSyncReactor", () => { ), ); + it.effect("syncs a pull request a reader saw merge without waiting for the sweep", () => + Effect.scoped( + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse(NOW)); + const state = yield* Ref.make("open"); + const invalidated = yield* Ref.make>([]); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([ + makeThread("agent", { pullRequests: [makeLink(7, { state: "open" })] }), + ]), + summary: (input) => + Ref.get(state).pipe( + Effect.map((current) => + makeSummary(input, current === "merged" ? { state: current, mergedAt: NOW } : {}), + ), + ), + invalidate: ({ reference }) => + Ref.update(invalidated, (numbers) => [...numbers, reference?.number ?? -1]), + }); + + yield* Effect.gen(function* () { + const reactor = yield* startAndSweep(fixture); + yield* Ref.set(state, "merged"); + + // The clock stays put: the next sweep is still a minute away. + yield* Queue.offer(fixture.stateChanges, { + host: "github.com", + repository: "owner/repository", + number: 7, + }); + yield* Queue.take(fixture.snapshotReads); + yield* reactor.drain; + + assert.deepStrictEqual(yield* Ref.get(invalidated), [7]); + assert.deepStrictEqual( + (yield* Ref.get(fixture.syncCommands)).map((command) => command.snapshot.state), + ["merged"], + ); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + it.effect("discovers externally reopened pull requests after fifteen minutes", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts index 5af4138bda9e428053d1a8af3038c4fbca4b19f1..d171e4b301c117caf3b8c807e47f3063a77ba4ed 100644 --- a/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts +++ b/apps/server/src/orchestration-v2/PullRequestSyncReactor.ts @@ -424,6 +424,13 @@ export const make = Effect.gen(function* () { "PullRequestSyncReactor.start", )(function* () { const events = engine.streamDomainEvents; + // A client reading a pull request can see it merge or close before the next sweep does. + const stateChanges = yield* pullRequests.subscribeStateChanges; + yield* forkParked( + Stream.runForEach(stateChanges, requestSync).pipe( + Effect.catchCause(logSkipped("pull request state change stream failed", {})), + ), + ); yield* forkParked( Stream.runForEach(events, (event) => { switch (event.type) { diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index d8fd366916ab14fd20a015754b0f662b6fe8c086..c2a46ed8897804c28e7d4c6b3fed68aec0ff45b2 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -4864,6 +4864,62 @@ it.effect("reuses an observed merged state for strict settlement reads", () => }), ); +it.effect("announces state a detail read sees first or newly", () => + Effect.gen(function* () { + let detail = { + state: "open" as "open" | "closed" | "merged", + updatedAt: "2026-07-02T00:00:00Z", + }; + let summaryState: "open" | "merged" = "open"; + const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; + const service = yield* makeService({ + projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })], + providers: [ + fakeProvider("github", { + getChangeRequest: () => Effect.succeed({ ...hostedChangeRequest("body"), ...detail }), + getChangeRequestSummary: () => + Effect.succeed({ + ...changeRequest(1, "2026-07-04T00:00:00Z"), + state: summaryState, + }), + }), + ], + }); + const announced: Array = []; + yield* Stream.runForEach(yield* service.subscribeStateChanges, (key) => + Effect.sync(() => announced.push(`${key.host}/${key.repository}#${key.number}`)), + ).pipe(Effect.forkChild({ startImmediately: true })); + const readDetail = Effect.gen(function* () { + yield* service.invalidate({ reference }); + yield* service.detail(reference); + yield* Effect.yieldNow; + }); + + // First sight announces; the same state again, even from a fresh read, does not. + yield* readDetail; + yield* readDetail; + assert.deepStrictEqual(announced, ["github.com/acme/web#1"]); + + detail = { state: "closed", updatedAt: "2026-07-03T00:00:00Z" }; + yield* readDetail; + assert.strictEqual(announced.length, 2); + + // A summary seeing the merge first does not hide it from the detail read after. + summaryState = "merged"; + yield* service.summary(reference, { recoverTransientFailure: false }); + yield* Effect.yieldNow; + assert.strictEqual(announced.length, 2); + detail = { state: "merged", updatedAt: "2026-07-04T00:00:00Z" }; + yield* readDetail; + assert.strictEqual(announced.length, 3); + + // A detail read older than the merge cannot announce a reopen. + detail = { state: "open", updatedAt: "2026-07-03T00:00:00Z" }; + yield* readDetail; + assert.strictEqual(announced.length, 3); + }), +); + it.effect("does not let a stale detail reopen overwrite a fresher linked summary", () => Effect.gen(function* () { const gate = yield* Deferred.make(); diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index 1e65926e84f4ebe8c90a02b058970082550640d6..3da0b9ba2e03b40d173e33858c6e3456294dff43 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -72,6 +72,7 @@ import { type PullRequestUpdateInput, type SourceControlProviderInfo, type SourceControlProviderKind, + type ThreadPullRequestKey, } from "@t3tools/contracts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { detectSourceControlProviderFromRemoteUrl } from "@t3tools/shared/sourceControl"; @@ -98,6 +99,16 @@ export interface PullRequestMergeEvent extends PullRequestRef { readonly mergedAt: string; } +type PullRequestReading = Pick; + +/** Whether `next` is a newer reading of a pull request than `current`. Merged is final. */ +function supersedesReading(current: PullRequestReading | undefined, next: PullRequestReading) { + if (current === undefined) return true; + if (current.state === "merged" && next.state !== "merged") return false; + if (next.updatedAt !== current.updatedAt) return next.updatedAt > current.updatedAt; + return (next.observedAt ?? -Infinity) >= (current.observedAt ?? -Infinity); +} + /** * Rows per repository when the client does not ask for a page size, and rows per slice when a * listing is carried on from a cursor. @@ -225,6 +236,15 @@ export class PullRequestService extends Context.Service< never, Scope.Scope >; + /** + * Pull requests a detail read saw for the first time or in a new state (opened, closed, + * merged), so thread links can catch up before the next sync sweep. Best effort. + */ + readonly subscribeStateChanges: Effect.Effect< + Stream.Stream, + never, + Scope.Scope + >; readonly subscribeRefreshes: Stream.Stream; readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect; readonly detail: (input: PullRequestRef) => Effect.Effect; @@ -666,6 +686,7 @@ const observeRead = Effect.fnUntraced(function* (read: Effect.Effect(64); + const stateChanges = yield* PubSub.sliding(64); const pullRequestRefreshes = yield* SubscriptionRef.make(0); const registry = yield* PullRequestProviderRegistry.PullRequestProviderRegistry; const serverSettings = yield* ServerSettings.ServerSettingsService; @@ -3057,13 +3078,39 @@ export const make = Effect.gen(function* () { updatedAt: detail.updatedAt, observedAt: detail.observedAt, }); - const shouldReplaceHeldSummary = (key: string, next: PullRequestSummary) => { - const current = lastGoodSummary.peek(key); - if (current === undefined) return true; - if (current.state === "merged" && next.state !== "merged") return false; - if (next.updatedAt !== current.updatedAt) return next.updatedAt > current.updatedAt; - return (next.observedAt ?? -Infinity) >= (current.observedAt ?? -Infinity); - }; + const shouldReplaceHeldSummary = (key: string, next: PullRequestSummary) => + supersedesReading(lastGoodSummary.peek(key), next); + // The newest state a detail read found per pull request, whatever cache epoch or credential + // it read under. Summary reads stay out: they would hide a change from the detail read after. + const detailStates = new Map(); + /** + * Announces a pull request a detail read sees first or in a new state, so a client showing a + * fresh merge brings the thread's link and its settlement along, rather than leaving them to + * the next sync sweep. + */ + const noteDetailReading = (ref: PullRequestRef, next: PullRequestSummary) => + Effect.suspend(() => { + const scope = refScope(ref); + const current = detailStates.get(scope); + if (!supersedesReading(current, next)) return Effect.void; + detailStates.delete(scope); + if (detailStates.size >= REF_EPOCH_CAPACITY) { + const oldest = detailStates.keys().next().value; + if (oldest !== undefined) detailStates.delete(oldest); + } + detailStates.set(scope, { + state: next.state, + updatedAt: next.updatedAt, + observedAt: next.observedAt, + }); + return current?.state !== next.state && ref.host !== undefined + ? PubSub.publish(stateChanges, { + host: ref.host, + repository: ref.repository, + number: ref.number, + }) + : Effect.void; + }); const detail: PullRequestService["Service"]["detail"] = (input) => { const key = refCacheKey(input); // Record the summary from a host or cache read, not the stale value @@ -3073,9 +3120,13 @@ export const make = Effect.gen(function* () { const read = Cache.get(detailCache, key).pipe( Effect.tap((value) => { const summary = summaryFromDetail(value, lastGoodSummary.peek(key)); - return shouldReplaceHeldSummary(key, summary) - ? lastGoodSummary.record(key, summary) - : Effect.void; + return noteDetailReading(input, summary).pipe( + Effect.andThen( + shouldReplaceHeldSummary(key, summary) + ? lastGoodSummary.record(key, summary) + : Effect.void, + ), + ); }), ); return input.allowStale === false @@ -3374,6 +3425,9 @@ export const make = Effect.gen(function* () { subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe( Effect.map((subscription) => Stream.fromSubscription(subscription)), ), + subscribeStateChanges: PubSub.subscribe(stateChanges).pipe( + Effect.map((subscription) => Stream.fromSubscription(subscription)), + ), subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe( Stream.filter((revision) => revision > 0), ),