diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index b4e04fd44f60..ce34855a3194 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -86,6 +86,7 @@ import { VcsStatusBroadcaster } from "../src/vcs/VcsStatusBroadcaster.ts"; import { GitWorkflowService } from "../src/git/GitWorkflowService.ts"; import * as VcsProcess from "../src/vcs/VcsProcess.ts"; import * as AgentAwarenessRelay from "../src/relay/AgentAwarenessRelay.ts"; +import * as PullRequestService from "../src/pullRequest/PullRequestService.ts"; const decodeCodexSettings = Schema.decodeEffect(CodexSettings); @@ -350,6 +351,11 @@ export const makeOrchestrationIntegrationHarness = ( ); const checkpointReactorLayer = CheckpointReactorLive.pipe( Layer.provideMerge(runtimeServicesLayer), + Layer.provideMerge( + Layer.mock(PullRequestService.PullRequestService)({ + refreshAfterTurn: Effect.void, + }), + ), Layer.provideMerge( Layer.succeed(VcsStatusBroadcaster, { getStatus: () => Effect.die("getStatus should not be called in this test"), diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 50adf83bfc53..de6661f45886 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -82,6 +82,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.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. [WS_METHODS.pullRequestsReviewerCandidates]: AuthOrchestrationReadScope, diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 87a239b13d92..391602548774 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -62,6 +62,7 @@ import { ProviderValidationError } from "../../provider/Errors.ts"; import { ServerConfig } from "../../config.ts"; import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts"; import * as WorkspacePaths from "../../workspace/WorkspacePaths.ts"; +import { PullRequestService } from "../../pullRequest/PullRequestService.ts"; const asProjectId = (value: string): ProjectId => ProjectId.make(value); const asTurnId = (value: string): TurnId => TurnId.make(value); @@ -325,6 +326,8 @@ describe("CheckpointReactor", () => { const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { prefix: "t3-checkpoint-reactor-test-", }); + const pullRequestRefreshes: number[] = []; + const refreshAfterTurn = Effect.sync(() => void pullRequestRefreshes.push(1)); const vcsStatusBroadcasterLayer = Layer.succeed(VcsStatusBroadcaster, { getStatus: () => Effect.die("getStatus should not be called in this test"), refreshLocalStatus: (cwd: string) => @@ -355,6 +358,7 @@ describe("CheckpointReactor", () => { Layer.provideMerge(projectionSnapshotLayer), Layer.provideMerge(RuntimeReceiptBusLive), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), + Layer.provideMerge(Layer.mock(PullRequestService)({ refreshAfterTurn })), Layer.provideMerge(vcsStatusBroadcasterLayer), Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))), Layer.provideMerge( @@ -466,6 +470,7 @@ describe("CheckpointReactor", () => { provider, cwd, drain, + pullRequestRefreshes, }; } @@ -766,6 +771,14 @@ describe("CheckpointReactor", () => { NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "v2\n", "utf8"); + harness.provider.emit({ + type: "turn.started", + eventId: EventId.make("evt-turn-started-aux"), + provider: ProviderDriverKind.make("codex"), + createdAt, + threadId: ThreadId.make("thread-1"), + turnId: asTurnId("turn-aux"), + }); harness.provider.emit({ type: "turn.completed", eventId: EventId.make("evt-turn-completed-aux"), @@ -782,6 +795,7 @@ describe("CheckpointReactor", () => { const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1")); expect(midThread?.checkpoints).toHaveLength(0); expect(pullRequestRefreshCalls).toEqual([]); + expect(harness.pullRequestRefreshes).toEqual([]); harness.provider.emit({ type: "turn.completed", @@ -801,6 +815,7 @@ describe("CheckpointReactor", () => { expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1); await harness.drain(); expect(pullRequestRefreshCalls).toEqual([harness.cwd]); + expect(harness.pullRequestRefreshes).toEqual([1]); }); it("captures pre-turn and completion checkpoints for claude runtime events", async () => { diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 0fc6295d4495..aa0e233a5dc3 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -38,6 +38,7 @@ import type { OrchestrationDispatchError } from "../Errors.ts"; import { isGitRepository } from "../../git/Utils.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts"; +import * as PullRequestService from "../../pullRequest/PullRequestService.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); @@ -88,6 +89,9 @@ const make = Effect.gen(function* () { const receiptBus = yield* RuntimeReceiptBus; const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; + const pullRequests = yield* PullRequestService.PullRequestService; + const startedTurns = new Map(); + const pending = new Set(); const appendRevertFailureActivity = (input: { readonly threadId: ThreadId; @@ -854,6 +858,7 @@ const make = Effect.gen(function* () { const processDomainEvent = Effect.fn("processDomainEvent")(function* (event: OrchestrationEvent) { if (event.type === "thread.turn-start-requested" || event.type === "thread.message-sent") { + if (event.type === "thread.turn-start-requested") pending.add(event.payload.threadId); yield* ensurePreTurnBaselineFromDomainTurnStart(event); return; } @@ -897,14 +902,46 @@ const make = Effect.gen(function* () { const processRuntimeEvent = Effect.fn("processRuntimeEvent")(function* ( event: ProviderRuntimeEvent, ) { + if (event.type === "session.exited") { + startedTurns.delete(event.threadId); + pending.delete(event.threadId); + return; + } + if (event.type === "turn.started") { + const turnId = toTurnId(event.turnId); + const activeTurnId = (yield* providerService.listSessions()).find((session) => + sameId(session.threadId, event.threadId), + )?.activeTurnId; + const mayReplace = pending.has(event.threadId) && sameId(activeTurnId, turnId); + if (turnId !== null && (!startedTurns.has(event.threadId) || mayReplace)) { + startedTurns.set(event.threadId, turnId); + pending.delete(event.threadId); + } yield* ensurePreTurnBaselineFromTurnStart(event); return; } - if (event.type === "turn.completed") { + if (event.type === "turn.completed" || event.type === "turn.aborted") { const turnId = toTurnId(event.turnId); - yield* refreshLocalGitStatusFromTurnCompletion(event); + const thread = yield* resolveThreadDetail(event.threadId); + const startedTurnId = startedTurns.get(event.threadId); + const isTrackedTurn = sameId(startedTurnId, turnId); + if (isTrackedTurn) startedTurns.delete(event.threadId); + if (event.type === "turn.completed") { + yield* refreshLocalGitStatusFromTurnCompletion(event); + } + if ( + turnId !== null && + thread !== undefined && + (isTrackedTurn || + sameId(thread.session?.activeTurnId, turnId) || + (startedTurnId === undefined && !thread.session?.activeTurnId)) + ) { + pending.delete(event.threadId); + yield* pullRequests.refreshAfterTurn; + } + if (event.type === "turn.aborted") return; yield* captureCheckpointFromTurnCompletion(event).pipe( Effect.catch((error) => Effect.flatMap(nowIso, (createdAt) => @@ -963,7 +1000,12 @@ const make = Effect.gen(function* () { yield* forkParked( Stream.runForEach(providerService.streamEvents, (event) => { - if (event.type !== "turn.started" && event.type !== "turn.completed") { + if ( + event.type !== "turn.started" && + event.type !== "turn.completed" && + event.type !== "turn.aborted" && + event.type !== "session.exited" + ) { return Effect.void; } return worker.enqueue({ source: "runtime", event }); diff --git a/apps/server/src/pullRequest/PullRequestService.test.ts b/apps/server/src/pullRequest/PullRequestService.test.ts index 0dc7a928c264..1f2ff59a94d3 100644 --- a/apps/server/src/pullRequest/PullRequestService.test.ts +++ b/apps/server/src/pullRequest/PullRequestService.test.ts @@ -2646,10 +2646,11 @@ it.effect("a listing narrowed to some projects is its own cache entry", () => }), ); -it.effect("an explicit invalidation makes the next listing ask the host again", () => +it.effect("explicit and turn invalidations make the next listing ask the host again", () => Effect.gen(function* () { let hostCalls = 0; let viewerCalls = 0; + 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: [ @@ -2673,11 +2674,14 @@ it.effect("an explicit invalidation makes the next listing ask the host again", assert.strictEqual(viewerCalls, 2); // Forgetting one change request leaves the listings shared. - yield* service.invalidate({ - reference: { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }, - }); + yield* service.invalidate({ reference }); yield* service.list({ state: "open" }); assert.strictEqual(hostCalls, 2); + yield* service.refreshAfterTurn; + const refresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes)); + yield* service.list({ state: "open" }); + assert.isAbove(refresh, 0); + assert.strictEqual(hostCalls, 3); }), ); @@ -3900,7 +3904,7 @@ it.effect("refuses a remark rewritten into nothing but whitespace", () => }), ); -it.effect("forgets the cached detail after a rewrite, like the other mutations", () => +it.effect("forgets the cached detail after a rewrite or terminal turn", () => Effect.gen(function* () { let coreCalls = 0; const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 }; @@ -3935,8 +3939,11 @@ it.effect("forgets the cached detail after a rewrite, like the other mutations", yield* service.detail(reference); yield* service.update({ ...reference, title: "Renamed" }); yield* service.detail(reference); - assert.strictEqual(coreCalls, 2); + + yield* service.refreshAfterTurn; + yield* service.detail(reference); + assert.strictEqual(coreCalls, 3); }), ); diff --git a/apps/server/src/pullRequest/PullRequestService.ts b/apps/server/src/pullRequest/PullRequestService.ts index b27fff3534a7..88a8ffe32df3 100644 --- a/apps/server/src/pullRequest/PullRequestService.ts +++ b/apps/server/src/pullRequest/PullRequestService.ts @@ -10,6 +10,7 @@ import * as PubSub from "effect/PubSub"; import * as Schema from "effect/Schema"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as SubscriptionRef from "effect/SubscriptionRef"; import { PullRequestOperationError, PullRequestUnavailableError, @@ -148,6 +149,8 @@ export class PullRequestService extends Context.Service< never, Scope.Scope >; + readonly subscribeRefreshes: Stream.Stream; + readonly refreshAfterTurn: Effect.Effect; readonly detail: (input: PullRequestRef) => Effect.Effect; readonly activity: ( input: PullRequestRef, @@ -531,6 +534,7 @@ export function repositoryIdentityOf(project: OrchestrationProjectShell): string export const make = Effect.gen(function* () { const mergedPullRequests = yield* PubSub.sliding(64); + const pullRequestRefreshes = yield* SubscriptionRef.make(0); const registry = yield* PullRequestProviderRegistry; const projections = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const sourceControlProviders = yield* SourceControlProviderRegistry.SourceControlProviderRegistry; @@ -2123,10 +2127,12 @@ export const make = Effect.gen(function* () { // scope re-entering `refEpochs` after eviction can never mint a key an old entry still has. let epochCounter = 0; let listingsEpoch = 0; + let turnRefreshEpoch = 0; const refEpochs = new Map(); const REF_EPOCH_CAPACITY = 2_048; const refScope = (ref: PullRequestRef) => `${ref.projectId} ${ref.repository} ${ref.number}`; - const refEpoch = (ref: PullRequestRef) => refEpochs.get(refScope(ref)) ?? 0; + const refEpoch = (ref: PullRequestRef) => + Math.max(turnRefreshEpoch, refEpochs.get(refScope(ref)) ?? 0); const refCacheKey = (ref: PullRequestRef) => JSON.stringify([refEpoch(ref), ref.projectId, ref.repository, ref.number]); const bumpRefEpoch = (ref: PullRequestRef) => { @@ -2396,6 +2402,11 @@ export const make = Effect.gen(function* () { }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights))); }; + const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => { + turnRefreshEpoch = listingsEpoch = ++epochCounter; + return SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch); + }); + // A mutation's own client re-reads right after it, and every other client's next read must // see the action too — so a write forgets the change request it touched and the listings its // state change reorders, for everyone, without any client asking. @@ -2435,6 +2446,10 @@ export const make = Effect.gen(function* () { subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe( Effect.map((subscription) => Stream.fromSubscription(subscription)), ), + subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe( + Stream.filter((revision) => revision > 0), + ), + refreshAfterTurn, detail, activity, threadComments, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0a9b5b64389d..35f41d58e319 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2096,6 +2096,12 @@ const makeWsRpcLayer = ( observeRpcEffect(WS_METHODS.pullRequestsInvalidate, pullRequests.invalidate(input), { "rpc.aggregate": "pull-requests", }), + [WS_METHODS.pullRequestsSubscribeRefreshes]: () => + observeRpcStream( + WS_METHODS.pullRequestsSubscribeRefreshes, + pullRequests.subscribeRefreshes, + { "rpc.aggregate": "pull-requests" }, + ), [WS_METHODS.pullRequestsReviewerCandidates]: (input) => observeRpcEffect( WS_METHODS.pullRequestsReviewerCandidates, diff --git a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx index adce43dd3344..9bfc49848803 100644 --- a/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx +++ b/apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx @@ -64,7 +64,11 @@ import { useProjects } from "~/state/entities"; import { useEnvironments } from "~/state/environments"; import { useEnvironmentQuery } from "~/state/query"; import { useLiveRefresh } from "~/hooks/useLiveRefresh"; -import { pullRequestEnvironment, useSharedPullRequestSummary } from "~/state/pullRequests"; +import { + pullRequestEnvironment, + usePullRequestTurnRefresh, + useSharedPullRequestSummary, +} from "~/state/pullRequests"; import { useAtomCommand } from "~/state/use-atom-command"; import { vcsEnvironment } from "~/state/vcs"; import { formatRelativeTimeLabel } from "~/timestampFormat"; @@ -577,6 +581,7 @@ export function PullRequestDetailPanel({ const activityQuery = useEnvironmentQuery( pullRequestEnvironment.activity({ environmentId, input: reference }), ); + const turnRefresh = usePullRequestTurnRefresh(environmentId); const [cachedDetail, setCachedDetail] = useState(() => readPullRequestDetailSnapshot( typeof window === "undefined" ? undefined : window.localStorage, @@ -675,6 +680,8 @@ export function PullRequestDetailPanel({ detailQuery.refresh(); activityQuery.refresh(); }, [activityQuery.refresh, detailQuery.refresh]); + const [refreshToken, setRefreshToken] = useState(0); + const codeRefreshToken = refreshToken + (turnRefresh ?? 0); const activityRevision = useRef<{ readonly key: string; readonly updatedAt: string } | null>( null, ); @@ -698,15 +705,18 @@ export function PullRequestDetailPanel({ // revision effect above reads it only after this same pull request reports a change. Keyed by // the pull request rather than by the panel, because this one panel shows a different pull // request every time it is opened. - useLiveRefresh(detailQuery.refresh, { - key: `pull-request:${reference.projectId}:${reference.repository}#${reference.number}`, - }); + useLiveRefresh( + () => { + detailQuery.refresh(); + setRefreshToken((token) => token + 1); + }, + { key: `pull-request:${reference.projectId}:${reference.repository}#${reference.number}` }, + ); // The button, on the other hand, goes around the server's cache rather than through it: it is // the answer for a reader who can see that what they are looking at is behind. The // invalidation goes first so the re-reads miss that cache; if it fails, the reads still run // and at worst answer from it. const invalidate = useAtomCommand(pullRequestEnvironment.invalidate, { reportFailure: false }); - const [refreshToken, setRefreshToken] = useState(0); const refreshFromHost = useCallback(async () => { await invalidate({ environmentId, input: { reference } }); refreshDetail(); @@ -2352,7 +2362,7 @@ export function PullRequestDetailPanel({ fixFindingLabel={handoffLabels.fixFinding} onFixFinding={startFixFinding} onRefresh={refreshDetail} - refreshToken={refreshToken} + refreshToken={codeRefreshToken} /> diff --git a/apps/web/src/routes/_chat.pull-requests.tsx b/apps/web/src/routes/_chat.pull-requests.tsx index 6db4f99b1502..bcc22daa4a9d 100644 --- a/apps/web/src/routes/_chat.pull-requests.tsx +++ b/apps/web/src/routes/_chat.pull-requests.tsx @@ -135,6 +135,7 @@ import { pullRequestEnvironment, usePullRequestList, usePullRequestListStats, + usePullRequestTurnRefreshes, type EnvironmentQueryTarget, } from "../state/pullRequests"; import { useAtomCommand } from "../state/use-atom-command"; @@ -632,6 +633,12 @@ function PullRequestsRouteView() { .join("|"), [environmentQueries], ); + const turnRefreshes = usePullRequestTurnRefreshes( + environmentQueries.map(({ environmentId }) => environmentId), + ); + const turnRefreshToken = turnRefreshes + .map(([environmentId, revision]) => `${environmentId}:${revision}`) + .join("|"); // Page size is view state, not a URL concern: a shared link should open the first page. const scopeKey = `${environmentKey}:${assignmentKey}:${search.state}:${search.involvement}:${scopedProjectId ?? ""}:${search.host ?? ""}:${search.draft ?? ""}:${search.review ?? ""}:${search.checks ?? ""}:${search.author ?? ""}:${search.labels?.join("\u0000") ?? ""}`; const filterKey = `${scopeKey}:${sentQuery}`; @@ -1091,6 +1098,18 @@ function PullRequestsRouteView() { }); }; + const appliedTurnRefreshToken = useRef(""); + const refreshAfterTurn = useEffectEvent(() => { + if (sentCursors !== null) refreshList(); + }); + useEffect(() => { + if (turnRefreshToken.length === 0 || appliedTurnRefreshToken.current === turnRefreshToken) { + return; + } + appliedTurnRefreshToken.current = turnRefreshToken; + refreshAfterTurn(); + }, [turnRefreshToken]); + // The list goes stale the same way the detail does: somebody opens a pull request, a check // finishes, a branch is merged. So it reads again on the way back to the window, and once a // minute while somebody is reading it. Those reads go through the server's cache and stop diff --git a/apps/web/src/state/pullRequests.ts b/apps/web/src/state/pullRequests.ts index bde8b4c2d9cf..99b2ef37c99a 100644 --- a/apps/web/src/state/pullRequests.ts +++ b/apps/web/src/state/pullRequests.ts @@ -24,8 +24,10 @@ import { import { formatEnvironmentQueryError } from "./query"; export const pullRequestEnvironment = createPullRequestEnvironmentAtoms(connectionAtomRuntime); -export const linkedPullRequestDetailAtom = - createLinkedPullRequestSummaryAtomFamily(connectionAtomRuntime); +export const linkedPullRequestDetailAtom = createLinkedPullRequestSummaryAtomFamily( + connectionAtomRuntime, + pullRequestEnvironment.refreshes, +); const observedPullRequestSummaryAtom = Atom.family((key: string) => Atom.make(null).pipe( @@ -148,6 +150,25 @@ const usePullRequestStatsQuery = createMergedEnvironmentQuery( pullRequestEnvironment.listStats, ); +const usePullRequestTurnRefreshQuery = createMergedEnvironmentQuery( + "web-pull-requests:turn-refreshes", + ({ environmentId }: EnvironmentQueryTarget>>) => + pullRequestEnvironment.refreshes({ environmentId, input: {} }), +); + +export function usePullRequestTurnRefreshes( + environmentIds: ReadonlyArray, +): ReadonlyArray { + return usePullRequestTurnRefreshQuery( + environmentIds.map((environmentId) => ({ environmentId, input: {} })), + ).values; +} + +export function usePullRequestTurnRefresh(environmentId: EnvironmentId): number | null { + const result = useAtomValue(pullRequestEnvironment.refreshes({ environmentId, input: {} })); + return Option.getOrNull(AsyncResult.value(result)); +} + export interface MergedPullRequestListView { readonly data: MergedPullRequestList | null; readonly error: string | null; diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index 38eb3735ab8f..0d68d2b2d531 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -52,6 +52,7 @@ export type EnvironmentSubscriptionRpcTag = | typeof WS_METHODS.subscribePreviewEvents | typeof WS_METHODS.subscribeDiscoveredLocalServers | typeof WS_METHODS.subscribeResourceTelemetry + | typeof WS_METHODS.pullRequestsSubscribeRefreshes | typeof WS_METHODS.previewAutomationConnect | typeof WS_METHODS.subscribeVcsStatus | typeof WS_METHODS.terminalAttach; diff --git a/packages/client-runtime/src/state/pullRequests.test.ts b/packages/client-runtime/src/state/pullRequests.test.ts index 618d5c39418b..6a22b22f1bda 100644 --- a/packages/client-runtime/src/state/pullRequests.test.ts +++ b/packages/client-runtime/src/state/pullRequests.test.ts @@ -1,8 +1,10 @@ import { EnvironmentId, ProjectId, WS_METHODS } from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Latch from "effect/Latch"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; @@ -42,8 +44,10 @@ function session(client: WsRpcProtocolClient): RpcSession { it.effect("refreshes pull request activity after a comment is updated", () => Effect.scoped( Effect.gen(function* () { + const refreshEvents = yield* PubSub.unbounded(); let commentBody = "old comment"; const client = { + [WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.fromPubSub(refreshEvents), [WS_METHODS.pullRequestsActivity]: () => Effect.succeed({ author: null, @@ -138,6 +142,22 @@ it.effect("refreshes pull request activity after a comment is updated", () => (yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).comments[0] ?.body, ).toBe("updated"); + const refreshed = Latch.makeUnsafe(); + const stop = registry.subscribe(activity, (result) => { + if (AsyncResult.isSuccess(result) && result.value.comments[0]?.body === "after turn") { + refreshed.openUnsafe(); + } + }); + yield* Effect.addFinalizer(() => Effect.sync(stop)); + + commentBody = "after turn"; + yield* PubSub.publish(refreshEvents, 1); + yield* refreshed.await; + + expect( + (yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).comments[0] + ?.body, + ).toBe("after turn"); }), ), ); diff --git a/packages/client-runtime/src/state/pullRequests.ts b/packages/client-runtime/src/state/pullRequests.ts index 28c85ffe5c1e..7ebaba0fed03 100644 --- a/packages/client-runtime/src/state/pullRequests.ts +++ b/packages/client-runtime/src/state/pullRequests.ts @@ -15,6 +15,7 @@ import { createAtomCommandScheduler, createEnvironmentRpcCommand, createEnvironmentRpcQueryAtomFamily, + createEnvironmentRpcSubscriptionAtomFamily, createEnvironmentQueryAtomFamily, } from "./runtime.ts"; import { PullRequestDiffLoader } from "./pullRequestDiffHttp.ts"; @@ -34,9 +35,19 @@ export class EnvironmentHttpConnectionNotReadyError extends Data.TaggedError( export const LINKED_PULL_REQUEST_IDLE_TTL_MS = 5_000; +function createPullRequestRefreshAtomFamily( + runtime: Atom.AtomRuntime, +) { + return createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:pull-requests:turn-refreshes", + tag: WS_METHODS.pullRequestsSubscribeRefreshes, + }); +} + /** Refresh only the live fields a linked thread renders. */ export function createLinkedPullRequestSummaryAtomFamily( runtime: Atom.AtomRuntime, + refreshes = createPullRequestRefreshAtomFamily(runtime), ) { return createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:pull-requests:linked-summary", @@ -44,6 +55,7 @@ export function createLinkedPullRequestSummaryAtomFamily( staleTimeMs: 60_000, refreshIntervalMs: 60_000, idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS, + refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), }); } @@ -69,6 +81,7 @@ export function pullRequestDetailToVcsStatus( export function createPullRequestEnvironmentAtoms( runtime: Atom.AtomRuntime, ) { + const refreshes = createPullRequestRefreshAtomFamily(runtime); const commandScheduler = createAtomCommandScheduler(); const serialPerEnvironment = { mode: "serial", @@ -78,12 +91,16 @@ export function createPullRequestEnvironmentAtoms( label: "environment-data:pull-requests:activity", tag: WS_METHODS.pullRequestsActivity, staleTimeMs: 15_000, + refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), }); return { + refreshes, list: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:pull-requests:list", tag: WS_METHODS.pullRequestsList, staleTimeMs: 30_000, + refreshTrigger: ({ environmentId, input }) => + input.cursors === undefined ? refreshes({ environmentId, input: {} }) : undefined, }), /** * The line counts for rows the listing has already handed over. Its own query because the @@ -95,11 +112,13 @@ export function createPullRequestEnvironmentAtoms( label: "environment-data:pull-requests:list-stats", tag: WS_METHODS.pullRequestsListStats, staleTimeMs: 60_000, + refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), }), detail: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:pull-requests:detail", tag: WS_METHODS.pullRequestsDetail, staleTimeMs: 15_000, + refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }), }), activity, threadComments: createEnvironmentRpcCommand(runtime, { diff --git a/packages/client-runtime/src/state/runtime.ts b/packages/client-runtime/src/state/runtime.ts index 4b0fc330839e..56489a4ba668 100644 --- a/packages/client-runtime/src/state/runtime.ts +++ b/packages/client-runtime/src/state/runtime.ts @@ -51,6 +51,10 @@ interface EnvironmentQueryAtomOptions extends EnvironmentAtomOpt readonly staleTimeMs?: number; readonly idleTtlMs?: number; readonly refreshIntervalMs?: number; + readonly refreshTrigger?: (target: { + readonly environmentId: EnvironmentIdType; + readonly input: Input; + }) => Atom.Atom | undefined; } interface EnvironmentSubscriptionAtomOptions { @@ -565,10 +569,15 @@ export function createEnvironmentQueryAtomFamily( }), Atom.setIdleTTL(idleTtlMs), ); - return ( + const intervalQuery = options.refreshIntervalMs === undefined ? queryAtom - : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs)) + : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs)); + const refreshTrigger = options.refreshTrigger?.(target); + return ( + refreshTrigger === undefined + ? intervalQuery + : intervalQuery.pipe(Atom.makeRefreshOnSignal(refreshTrigger)) ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`)); }); return (target) => family(environmentRpcKey(target)); @@ -615,6 +624,10 @@ export function createEnvironmentRpcQueryAtomFamily; + }) => Atom.Atom | undefined; }, ) { return createEnvironmentQueryAtomFamily(runtime, { @@ -624,6 +637,7 @@ export function createEnvironmentRpcQueryAtomFamily) => request(options.tag, input), }); } diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index c0ef8cd56d6d..f7f2c2b6faa7 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1,7 +1,7 @@ import * as Schema from "effect/Schema"; import * as Rpc from "effect/unstable/rpc/Rpc"; import * as RpcGroup from "effect/unstable/rpc/RpcGroup"; -import { TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; import { ProviderAuthCancelInput, ProviderAuthCompleteInput, @@ -343,6 +343,7 @@ export const WS_METHODS = { pullRequestsSetThreadResolution: "pullRequests.setThreadResolution", pullRequestsSetReaction: "pullRequests.setReaction", pullRequestsInvalidate: "pullRequests.invalidate", + pullRequestsSubscribeRefreshes: "pullRequests.subscribeRefreshes", pullRequestsReviewerCandidates: "pullRequests.reviewerCandidates", pullRequestsRequestReviewers: "pullRequests.requestReviewers", pullRequestsLabelCandidates: "pullRequests.labelCandidates", @@ -714,6 +715,16 @@ export const WsPullRequestsInvalidateRpc = Rpc.make(WS_METHODS.pullRequestsInval error: PullRequestRpcError, }); +export const WsPullRequestsSubscribeRefreshesRpc = Rpc.make( + WS_METHODS.pullRequestsSubscribeRefreshes, + { + payload: Schema.Struct({}), + success: NonNegativeInt, + error: EnvironmentAuthorizationError, + stream: true, + }, +); + /** * Read on its own rather than as part of the detail: the people who may be asked are only wanted * once somebody opens the menu, and reading them with every change request would spend a request @@ -1215,6 +1226,7 @@ export const WsRpcGroup = RpcGroup.make( WsPullRequestsSetThreadResolutionRpc, WsPullRequestsSetReactionRpc, WsPullRequestsInvalidateRpc, + WsPullRequestsSubscribeRefreshesRpc, WsPullRequestsReviewerCandidatesRpc, WsPullRequestsRequestReviewersRpc, WsPullRequestsLabelCandidatesRpc,