diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 511fd77cd14792d3c33551090ffb67277d596000..11756026f6d1b9013ef221c37af1c7217cd73905 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -135,6 +135,7 @@ export const RPC_REQUIRED_SCOPES = { // Read scope like the reads it un-caches: refreshing is part of reading, and a read-only // client pressing refresh must not be told it may not look again. [WS_METHODS.pullRequestsInvalidate]: AuthOrchestrationReadScope, + [WS_METHODS.pullRequestsReportState]: AuthOrchestrationReadScope, [WS_METHODS.pullRequestsSubscribeRefreshes]: AuthOrchestrationReadScope, // The candidate list is a read like the detail beside it; asking somebody for a review is a // write like every other one. diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 2ad903ff792ab69838c5dcc1cb384f2d74b60d12..26a4b007b8b66163bf41abb8a0918f9cf5969876 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -118,6 +118,7 @@ const RPC_AGGREGATES = { [WS_METHODS.pullRequestsSetReaction]: "pull-requests", [WS_METHODS.pullRequestsSetFilesViewed]: "pull-requests", [WS_METHODS.pullRequestsInvalidate]: "pull-requests", + [WS_METHODS.pullRequestsReportState]: "pull-requests", [WS_METHODS.pullRequestsSubscribeRefreshes]: "pull-requests", [WS_METHODS.pullRequestsReviewerCandidates]: "pull-requests", [WS_METHODS.pullRequestsRequestReviewers]: "pull-requests", diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index c74faa92f06dc5629390f69fad926ed0875e6d7a..117fd3f69a3291d4d5fbaf5e7e59ffee72939dd6 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -4972,6 +4972,48 @@ it.effect("announces state a detail read sees first or newly", () => }), ); +it.effect("announces a reported state without trusting it", () => + Effect.gen(function* () { + let detail = { state: "open" as "open" | "closed", updatedAt: "2026-07-02T00:00:00Z" }; + 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 }), + }), + ], + }); + 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 note = (state: "open" | "closed" | "merged") => + service.reportState({ reference, state }).pipe(Effect.andThen(Effect.yieldNow)); + const readDetail = Effect.gen(function* () { + yield* service.invalidate({ reference }); + yield* service.detail(reference); + yield* Effect.yieldNow; + }); + + yield* readDetail; + yield* note("closed"); + yield* note("closed"); + assert.strictEqual(announced.length, 2); + yield* TestClock.adjust("5 minutes"); + yield* note("closed"); + assert.strictEqual(announced.length, 3); + + yield* note("open"); + assert.strictEqual(announced.length, 4); + + yield* note("merged"); + detail = { state: "closed", updatedAt: "2026-07-03T00:00:00Z" }; + yield* readDetail; + assert.strictEqual(announced.length, 6); + }), +); + 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 b44d713c3158896be8d9465b20476fa456aa11cb..51377f1b33b193c35ed16fed893936a0caf3b6d4 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -42,6 +42,7 @@ import { type PullRequestFilesViewedResult, type PullRequestDiffResult, type PullRequestInvalidateInput, + type PullRequestReportStateInput, type PullRequestListEntry, type PullRequestListFilters, type PullRequestListInput, @@ -160,6 +161,8 @@ const detailTimeToLive = (state: PullRequestState | undefined) => * page and a watched pull request's change reaches its watch. */ const CHECKS_CACHE_TTL = Duration.seconds(15); +/** How long a repeated reported state is skipped, so a wrong one cannot hide a real change for long. */ +const REPORTED_STATE_DEDUPE_MS = 5 * 60 * 1_000; const DIFF_CACHE_TTL = Duration.seconds(60); /** A commit is content-addressed, so its own diff cannot change under its key. */ const COMMIT_DIFF_CACHE_TTL = Duration.minutes(10); @@ -244,6 +247,13 @@ export class PullRequestService extends Context.Service< never, Scope.Scope >; + /** + * The state a read routed to another environment saw. A new one reaches + * `subscribeStateChanges`, so the host is asked again; the report itself is never trusted. + */ + readonly reportState: ( + input: PullRequestReportStateInput, + ) => Effect.Effect; readonly subscribeRefreshes: Stream.Stream; readonly refreshAfterTurn: (projectId: ProjectId) => Effect.Effect; readonly detail: (input: PullRequestRef) => Effect.Effect; @@ -3115,6 +3125,36 @@ export const make = Effect.gen(function* () { }) : Effect.void; }); + // The last state each routed read reported, kept apart from `detailStates`: an unverified report + // must never outrank a reading this environment made itself. + const reportedStates = new Map< + string, + { readonly state: PullRequestState; readonly atMs: number } + >(); + const reportState: PullRequestService["Service"]["reportState"] = ({ reference, state }) => + Effect.all([canonicalRef(reference), Clock.currentTimeMillis]).pipe( + Effect.flatMap(([ref, nowMs]) => + Effect.suspend(() => { + const scope = refScope(ref); + const reported = reportedStates.get(scope); + if (reported?.state === state && nowMs - reported.atMs < REPORTED_STATE_DEDUPE_MS) + return Effect.void; + reportedStates.delete(scope); + if (reportedStates.size >= REF_EPOCH_CAPACITY) { + const oldest = reportedStates.keys().next().value; + if (oldest !== undefined) reportedStates.delete(oldest); + } + reportedStates.set(scope, { state, atMs: nowMs }); + return ref.host === undefined + ? Effect.void + : PubSub.publish(stateChanges, { + host: ref.host, + repository: ref.repository, + number: ref.number, + }); + }), + ), + ); const detail: PullRequestService["Service"]["detail"] = (input) => { const key = refCacheKey(input); // Record the summary from a host or cache read, not the stale value @@ -3432,6 +3472,7 @@ export const make = Effect.gen(function* () { subscribeStateChanges: PubSub.subscribe(stateChanges).pipe( Effect.map((subscription) => Stream.fromSubscription(subscription)), ), + reportState, subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe( Stream.filter((revision) => revision > 0), ), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index fce520ca7e3b58e332ae94afeb501b3fd9af6c9c..c1a01b963a508deb760a32649f8018340a4510fd 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2518,6 +2518,7 @@ const layerWsRpc = ( withPullRequestViewer(input, pullRequests.setThreadResolution(input)), [WS_METHODS.pullRequestsSetReaction]: (input) => withPullRequestViewer(input, pullRequests.setReaction(input)), + [WS_METHODS.pullRequestsReportState]: (input) => pullRequests.reportState(input), [WS_METHODS.pullRequestsInvalidate]: (input) => pullRequests.invalidate(input, { notifyReaders: true }).pipe( // A reader asking for fresh host state also wants the thread badges it feeds to diff --git a/packages/client-runtime/src/state/pullRequestRouting.ts b/packages/client-runtime/src/state/pullRequestRouting.ts index f8781837c36191e29cd14cca7d5a8f87c46477e5..2233dd8a6722a53cff7cffba910ae0df9cc90e92 100644 --- a/packages/client-runtime/src/state/pullRequestRouting.ts +++ b/packages/client-runtime/src/state/pullRequestRouting.ts @@ -2,6 +2,7 @@ import { WS_METHODS, PullRequestRef, PullRequestInvalidateInput, + PullRequestState, type EnvironmentId, type PullRequestRoutingResult, type PullRequestRoutingIdentityResult, @@ -57,6 +58,7 @@ const writes = new Set([ WS_METHODS.pullRequestsSetLabels, ]); const isRef = Schema.is(PullRequestRef); +const hasState = Schema.is(Schema.Struct({ state: PullRequestState })); const isInvalidation = Schema.is(PullRequestInvalidateInput); const readTimeout = (environmentId: EnvironmentId) => Effect.timeoutOrElse({ @@ -208,6 +210,21 @@ const invalidateTarget = Effect.fn("PullRequestRouting.invalidateTarget")(functi ); }); +/** + * The origin owns the threads linked to this pull request, but a routed read never reaches it. + * Tell it which state the read saw so it checks again. Best effort, and never holds up the read. + */ +const reportState = ( + { allowStale: _allowStale, expectedAccountId: _expectedAccountId, ...reference }: PullRequestRef, + state: PullRequestState, +) => + request(WS_METHODS.pullRequestsReportState, { reference, state }).pipe( + Effect.timeoutOption("5 seconds"), + Effect.ignore, + Effect.forkDetach, + Effect.asVoid, + ); + /** Credentials stay on their environments. Only a verified host and account cross the wire. */ export function createPullRequestRouter() { const routedRequest = Effect.fn("PullRequestRouting.request")(function* < @@ -356,6 +373,11 @@ export function createPullRequestRouter() { ? guardedSource : registry.run(id, request(tag, routedInput)) ).pipe( + Effect.tap((result) => + id !== origin.target.environmentId && hasState(result) + ? reportState(ref, result.state) + : Effect.void, + ), Effect.tap(() => Effect.sync(() => { const entry = used.get(refKey) ?? { diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 8762e9edb39996a76f5848c25e18f81c50905608..75b76d8c3f4ef11ac5dfbf2c83f1aacc8d3dea6b 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -995,6 +995,7 @@ it.effect.each( it.live("keeps source workspace metadata when an alternate answers a detail read", () => Effect.scoped( Effect.gen(function* () { + const noted = yield* Deferred.make(); const clientFor = (local: boolean) => ({ [local ? WS_METHODS.pullRequestsRoutingIdentity : WS_METHODS.pullRequestsRouting]: () => @@ -1013,8 +1014,12 @@ it.live("keeps source workspace metadata when an alternate answers a detail read projectTitle: "local project", workspaceRoot: "/Users/local/repo", title: "github title", + state: "closed", + updatedAt: "2026-10-10T01:52:36Z", }) : Effect.never, + [WS_METHODS.pullRequestsReportState]: (input: unknown) => + Deferred.succeed(noted, input).pipe(Effect.asVoid), }) as unknown as WsRpcProtocolClient; const { environmentRegistry, supervisor } = yield* makeTestRuntime( clientFor(false), @@ -1035,6 +1040,12 @@ it.live("keeps source workspace metadata when an alternate answers a detail read projectTitle: "source project", workspaceRoot: "/srv/source/repo", title: "github title", + state: "closed", + updatedAt: "2026-10-10T01:52:36Z", + }); + expect(yield* Deferred.await(noted)).toEqual({ + reference: { projectId: "project-1", repository: "acme/web", number: 7 }, + state: "closed", }); }), ), diff --git a/packages/contracts/src/pullRequest.ts b/packages/contracts/src/pullRequest.ts index 1efbc1808d143beddebc0934d3680fa4870a3a93..82e5dc1cd15f46ba697d6807c3baaf1d3dca0699 100644 --- a/packages/contracts/src/pullRequest.ts +++ b/packages/contracts/src/pullRequest.ts @@ -849,6 +849,17 @@ export const PullRequestInvalidateInput = Schema.Struct({ }); export type PullRequestInvalidateInput = typeof PullRequestInvalidateInput.Type; +/** + * The state a read routed to another environment saw, reported to the environment the read was + * for so its thread links can catch up. A hint only: that environment confirms with the host + * before writing anything. + */ +export const PullRequestReportStateInput = Schema.Struct({ + reference: PullRequestRef, + state: PullRequestState, +}); +export type PullRequestReportStateInput = typeof PullRequestReportStateInput.Type; + export const PullRequestDetail = Schema.Struct({ provider: SourceControlProviderKind, capabilities: PullRequestCapabilities, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 36afee2f9efa0f730bd17632403823dc5ac62891..289b7041be16e4bbf4f03045ae7c1b83e0953836 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -171,6 +171,7 @@ import { PullRequestDiffFileContentsResult, PullRequestFilesViewedResult, PullRequestInvalidateInput, + PullRequestReportStateInput, PullRequestListInput, PullRequestListResult, PullRequestListStatsInput, @@ -543,6 +544,7 @@ export const WS_METHODS = { pullRequestsSetThreadResolution: "pullRequests.setThreadResolution", pullRequestsSetReaction: "pullRequests.setReaction", pullRequestsInvalidate: "pullRequests.invalidate", + pullRequestsReportState: "pullRequests.reportState", pullRequestsSubscribeRefreshes: "pullRequests.subscribeRefreshes", pullRequestsReviewerCandidates: "pullRequests.reviewerCandidates", pullRequestsRequestReviewers: "pullRequests.requestReviewers", @@ -1087,6 +1089,12 @@ const WsPullRequestsInvalidateRpc = Rpc.make(WS_METHODS.pullRequestsInvalidate, error: PullRequestRpcError, }); +const WsPullRequestsReportStateRpc = Rpc.make(WS_METHODS.pullRequestsReportState, { + payload: PullRequestReportStateInput, + success: Schema.Void, + error: PullRequestRpcError, +}); + const WsPullRequestsSubscribeRefreshesRpc = Rpc.make(WS_METHODS.pullRequestsSubscribeRefreshes, { payload: Schema.Struct({}), success: NonNegativeInt, @@ -1916,6 +1924,7 @@ export const WsRpcGroup = RpcGroup.make( WsPullRequestsSetThreadResolutionRpc, WsPullRequestsSetReactionRpc, WsPullRequestsInvalidateRpc, + WsPullRequestsReportStateRpc, WsPullRequestsSubscribeRefreshesRpc, WsPullRequestsReviewerCandidatesRpc, WsPullRequestsRequestReviewersRpc,