authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-04 02:16:25-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 02:16:25-04:00
log5cc369b7eb882dbea8d5ad21ba688c73a0058748
tree943662732e652ba61d76075d7026d07804f1892b
parentf559fe0ba6fb5950bd14a2404f10b9c94b33f696
signature Signed by PGP key B5690EEEBB952194

fix(pull-requests): refresh data after thread turns (#9496)

Co-authored-by: maria-rcks <254055478+maria-rcks@users.noreply.github.com>

15 files changed, 229 insertions(+), 21 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+6
......@@ -86,6 +86,7 @@ import { VcsStatusBroadcaster } from "../src/vcs/VcsStatusBroadcaster.ts";
8686import { GitWorkflowService } from "../src/git/GitWorkflowService.ts";
8787import * as VcsProcess from "../src/vcs/VcsProcess.ts";
8888import * as AgentAwarenessRelay from "../src/relay/AgentAwarenessRelay.ts";
89import * as PullRequestService from "../src/pullRequest/PullRequestService.ts";
8990
9091const decodeCodexSettings = Schema.decodeEffect(CodexSettings);
9192
......@@ -350,6 +351,11 @@ export const makeOrchestrationIntegrationHarness = (
350351 );
351352 const checkpointReactorLayer = CheckpointReactorLive.pipe(
352353 Layer.provideMerge(runtimeServicesLayer),
354 Layer.provideMerge(
355 Layer.mock(PullRequestService.PullRequestService)({
356 refreshAfterTurn: Effect.void,
357 }),
358 ),
353359 Layer.provideMerge(
354360 Layer.succeed(VcsStatusBroadcaster, {
355361 getStatus: () => Effect.die("getStatus should not be called in this test"),
apps/server/src/auth/RpcAuthorization.ts+1
......@@ -82,6 +82,7 @@ export const RPC_REQUIRED_SCOPES = {
8282 // Read scope like the reads it un-caches: refreshing is part of reading, and a read-only
8383 // client pressing refresh must not be told it may not look again.
8484 [WS_METHODS.pullRequestsInvalidate]: AuthOrchestrationReadScope,
85 [WS_METHODS.pullRequestsSubscribeRefreshes]: AuthOrchestrationReadScope,
8586 // The candidate list is a read like the detail beside it; asking somebody for a review is a
8687 // write like every other one.
8788 [WS_METHODS.pullRequestsReviewerCandidates]: AuthOrchestrationReadScope,
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+15
......@@ -62,6 +62,7 @@ import { ProviderValidationError } from "../../provider/Errors.ts";
6262import { ServerConfig } from "../../config.ts";
6363import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts";
6464import * as WorkspacePaths from "../../workspace/WorkspacePaths.ts";
65import { PullRequestService } from "../../pullRequest/PullRequestService.ts";
6566
6667const asProjectId = (value: string): ProjectId => ProjectId.make(value);
6768const asTurnId = (value: string): TurnId => TurnId.make(value);
......@@ -325,6 +326,8 @@ describe("CheckpointReactor", () => {
325326 const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), {
326327 prefix: "t3-checkpoint-reactor-test-",
327328 });
329 const pullRequestRefreshes: number[] = [];
330 const refreshAfterTurn = Effect.sync(() => void pullRequestRefreshes.push(1));
328331 const vcsStatusBroadcasterLayer = Layer.succeed(VcsStatusBroadcaster, {
329332 getStatus: () => Effect.die("getStatus should not be called in this test"),
330333 refreshLocalStatus: (cwd: string) =>
......@@ -355,6 +358,7 @@ describe("CheckpointReactor", () => {
355358 Layer.provideMerge(projectionSnapshotLayer),
356359 Layer.provideMerge(RuntimeReceiptBusLive),
357360 Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
361 Layer.provideMerge(Layer.mock(PullRequestService)({ refreshAfterTurn })),
358362 Layer.provideMerge(vcsStatusBroadcasterLayer),
359363 Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))),
360364 Layer.provideMerge(
......@@ -466,6 +470,7 @@ describe("CheckpointReactor", () => {
466470 provider,
467471 cwd,
468472 drain,
473 pullRequestRefreshes,
469474 };
470475 }
471476
......@@ -766,6 +771,14 @@ describe("CheckpointReactor", () => {
766771
767772 NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "v2\n", "utf8");
768773
774 harness.provider.emit({
775 type: "turn.started",
776 eventId: EventId.make("evt-turn-started-aux"),
777 provider: ProviderDriverKind.make("codex"),
778 createdAt,
779 threadId: ThreadId.make("thread-1"),
780 turnId: asTurnId("turn-aux"),
781 });
769782 harness.provider.emit({
770783 type: "turn.completed",
771784 eventId: EventId.make("evt-turn-completed-aux"),
......@@ -782,6 +795,7 @@ describe("CheckpointReactor", () => {
782795 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
783796 expect(midThread?.checkpoints).toHaveLength(0);
784797 expect(pullRequestRefreshCalls).toEqual([]);
798 expect(harness.pullRequestRefreshes).toEqual([]);
785799
786800 harness.provider.emit({
787801 type: "turn.completed",
......@@ -801,6 +815,7 @@ describe("CheckpointReactor", () => {
801815 expect(thread.checkpoints[0]?.checkpointTurnCount).toBe(1);
802816 await harness.drain();
803817 expect(pullRequestRefreshCalls).toEqual([harness.cwd]);
818 expect(harness.pullRequestRefreshes).toEqual([1]);
804819 });
805820
806821 it("captures pre-turn and completion checkpoints for claude runtime events", async () => {
apps/server/src/orchestration/Layers/CheckpointReactor.ts+45-3
......@@ -38,6 +38,7 @@ import type { OrchestrationDispatchError } from "../Errors.ts";
3838import { isGitRepository } from "../../git/Utils.ts";
3939import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";
4040import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts";
41import * as PullRequestService from "../../pullRequest/PullRequestService.ts";
4142
4243const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
4344
......@@ -88,6 +89,9 @@ const make = Effect.gen(function* () {
8889 const receiptBus = yield* RuntimeReceiptBus;
8990 const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
9091 const vcsStatusBroadcaster = yield* VcsStatusBroadcaster;
92 const pullRequests = yield* PullRequestService.PullRequestService;
93 const startedTurns = new Map<ThreadId, TurnId>();
94 const pending = new Set<ThreadId>();
9195
9296 const appendRevertFailureActivity = (input: {
9397 readonly threadId: ThreadId;
......@@ -854,6 +858,7 @@ const make = Effect.gen(function* () {
854858
855859 const processDomainEvent = Effect.fn("processDomainEvent")(function* (event: OrchestrationEvent) {
856860 if (event.type === "thread.turn-start-requested" || event.type === "thread.message-sent") {
861 if (event.type === "thread.turn-start-requested") pending.add(event.payload.threadId);
857862 yield* ensurePreTurnBaselineFromDomainTurnStart(event);
858863 return;
859864 }
......@@ -897,14 +902,46 @@ const make = Effect.gen(function* () {
897902 const processRuntimeEvent = Effect.fn("processRuntimeEvent")(function* (
898903 event: ProviderRuntimeEvent,
899904 ) {
905 if (event.type === "session.exited") {
906 startedTurns.delete(event.threadId);
907 pending.delete(event.threadId);
908 return;
909 }
910
900911 if (event.type === "turn.started") {
912 const turnId = toTurnId(event.turnId);
913 const activeTurnId = (yield* providerService.listSessions()).find((session) =>
914 sameId(session.threadId, event.threadId),
915 )?.activeTurnId;
916 const mayReplace = pending.has(event.threadId) && sameId(activeTurnId, turnId);
917 if (turnId !== null && (!startedTurns.has(event.threadId) || mayReplace)) {
918 startedTurns.set(event.threadId, turnId);
919 pending.delete(event.threadId);
920 }
901921 yield* ensurePreTurnBaselineFromTurnStart(event);
902922 return;
903923 }
904924
905 if (event.type === "turn.completed") {
925 if (event.type === "turn.completed" || event.type === "turn.aborted") {
906926 const turnId = toTurnId(event.turnId);
907 yield* refreshLocalGitStatusFromTurnCompletion(event);
927 const thread = yield* resolveThreadDetail(event.threadId);
928 const startedTurnId = startedTurns.get(event.threadId);
929 const isTrackedTurn = sameId(startedTurnId, turnId);
930 if (isTrackedTurn) startedTurns.delete(event.threadId);
931 if (event.type === "turn.completed") {
932 yield* refreshLocalGitStatusFromTurnCompletion(event);
933 }
934 if (
935 turnId !== null &&
936 thread !== undefined &&
937 (isTrackedTurn ||
938 sameId(thread.session?.activeTurnId, turnId) ||
939 (startedTurnId === undefined && !thread.session?.activeTurnId))
940 ) {
941 pending.delete(event.threadId);
942 yield* pullRequests.refreshAfterTurn;
943 }
944 if (event.type === "turn.aborted") return;
908945 yield* captureCheckpointFromTurnCompletion(event).pipe(
909946 Effect.catch((error) =>
910947 Effect.flatMap(nowIso, (createdAt) =>
......@@ -963,7 +1000,12 @@ const make = Effect.gen(function* () {
9631000
9641001 yield* forkParked(
9651002 Stream.runForEach(providerService.streamEvents, (event) => {
966 if (event.type !== "turn.started" && event.type !== "turn.completed") {
1003 if (
1004 event.type !== "turn.started" &&
1005 event.type !== "turn.completed" &&
1006 event.type !== "turn.aborted" &&
1007 event.type !== "session.exited"
1008 ) {
9671009 return Effect.void;
9681010 }
9691011 return worker.enqueue({ source: "runtime", event });
apps/server/src/pullRequest/PullRequestService.test.ts+13-6
......@@ -2646,10 +2646,11 @@ it.effect("a listing narrowed to some projects is its own cache entry", () =>
26462646 }),
26472647);
26482648
2649it.effect("an explicit invalidation makes the next listing ask the host again", () =>
2649it.effect("explicit and turn invalidations make the next listing ask the host again", () =>
26502650 Effect.gen(function* () {
26512651 let hostCalls = 0;
26522652 let viewerCalls = 0;
2653 const reference = { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 };
26532654 const service = yield* makeService({
26542655 projects: [project({ id: "p1", title: "web", workspaceRoot: "/a", repository: "acme/web" })],
26552656 providers: [
......@@ -2673,11 +2674,14 @@ it.effect("an explicit invalidation makes the next listing ask the host again",
26732674 assert.strictEqual(viewerCalls, 2);
26742675
26752676 // Forgetting one change request leaves the listings shared.
2676 yield* service.invalidate({
2677 reference: { projectId: "p1" as ProjectId, repository: "acme/web", number: 1 },
2678 });
2677 yield* service.invalidate({ reference });
26792678 yield* service.list({ state: "open" });
26802679 assert.strictEqual(hostCalls, 2);
2680 yield* service.refreshAfterTurn;
2681 const refresh = Option.getOrThrow(yield* Stream.runHead(service.subscribeRefreshes));
2682 yield* service.list({ state: "open" });
2683 assert.isAbove(refresh, 0);
2684 assert.strictEqual(hostCalls, 3);
26812685 }),
26822686);
26832687
......@@ -3900,7 +3904,7 @@ it.effect("refuses a remark rewritten into nothing but whitespace", () =>
39003904 }),
39013905);
39023906
3903it.effect("forgets the cached detail after a rewrite, like the other mutations", () =>
3907it.effect("forgets the cached detail after a rewrite or terminal turn", () =>
39043908 Effect.gen(function* () {
39053909 let coreCalls = 0;
39063910 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",
39353939 yield* service.detail(reference);
39363940 yield* service.update({ ...reference, title: "Renamed" });
39373941 yield* service.detail(reference);
3938
39393942 assert.strictEqual(coreCalls, 2);
3943
3944 yield* service.refreshAfterTurn;
3945 yield* service.detail(reference);
3946 assert.strictEqual(coreCalls, 3);
39403947 }),
39413948);
39423949
apps/server/src/pullRequest/PullRequestService.ts+16-1
......@@ -10,6 +10,7 @@ import * as PubSub from "effect/PubSub";
1010import * as Schema from "effect/Schema";
1111import type * as Scope from "effect/Scope";
1212import * as Stream from "effect/Stream";
13import * as SubscriptionRef from "effect/SubscriptionRef";
1314import {
1415 PullRequestOperationError,
1516 PullRequestUnavailableError,
......@@ -148,6 +149,8 @@ export class PullRequestService extends Context.Service<
148149 never,
149150 Scope.Scope
150151 >;
152 readonly subscribeRefreshes: Stream.Stream<number>;
153 readonly refreshAfterTurn: Effect.Effect<void>;
151154 readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;
152155 readonly activity: (
153156 input: PullRequestRef,
......@@ -531,6 +534,7 @@ export function repositoryIdentityOf(project: OrchestrationProjectShell): string
531534
532535export const make = Effect.gen(function* () {
533536 const mergedPullRequests = yield* PubSub.sliding<PullRequestMergeEvent>(64);
537 const pullRequestRefreshes = yield* SubscriptionRef.make(0);
534538 const registry = yield* PullRequestProviderRegistry;
535539 const projections = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;
536540 const sourceControlProviders = yield* SourceControlProviderRegistry.SourceControlProviderRegistry;
......@@ -2123,10 +2127,12 @@ export const make = Effect.gen(function* () {
21232127 // scope re-entering `refEpochs` after eviction can never mint a key an old entry still has.
21242128 let epochCounter = 0;
21252129 let listingsEpoch = 0;
2130 let turnRefreshEpoch = 0;
21262131 const refEpochs = new Map<string, number>();
21272132 const REF_EPOCH_CAPACITY = 2_048;
21282133 const refScope = (ref: PullRequestRef) => `${ref.projectId} ${ref.repository} ${ref.number}`;
2129 const refEpoch = (ref: PullRequestRef) => refEpochs.get(refScope(ref)) ?? 0;
2134 const refEpoch = (ref: PullRequestRef) =>
2135 Math.max(turnRefreshEpoch, refEpochs.get(refScope(ref)) ?? 0);
21302136 const refCacheKey = (ref: PullRequestRef) =>
21312137 JSON.stringify([refEpoch(ref), ref.projectId, ref.repository, ref.number]);
21322138 const bumpRefEpoch = (ref: PullRequestRef) => {
......@@ -2396,6 +2402,11 @@ export const make = Effect.gen(function* () {
23962402 }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights)));
23972403 };
23982404
2405 const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.suspend(() => {
2406 turnRefreshEpoch = listingsEpoch = ++epochCounter;
2407 return SubscriptionRef.set(pullRequestRefreshes, turnRefreshEpoch);
2408 });
2409
23992410 // A mutation's own client re-reads right after it, and every other client's next read must
24002411 // see the action too — so a write forgets the change request it touched and the listings its
24012412 // state change reorders, for everyone, without any client asking.
......@@ -2435,6 +2446,10 @@ export const make = Effect.gen(function* () {
24352446 subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(
24362447 Effect.map((subscription) => Stream.fromSubscription(subscription)),
24372448 ),
2449 subscribeRefreshes: SubscriptionRef.changes(pullRequestRefreshes).pipe(
2450 Stream.filter((revision) => revision > 0),
2451 ),
2452 refreshAfterTurn,
24382453 detail,
24392454 activity,
24402455 threadComments,
apps/server/src/ws.ts+6
......@@ -2096,6 +2096,12 @@ const makeWsRpcLayer = (
20962096 observeRpcEffect(WS_METHODS.pullRequestsInvalidate, pullRequests.invalidate(input), {
20972097 "rpc.aggregate": "pull-requests",
20982098 }),
2099 [WS_METHODS.pullRequestsSubscribeRefreshes]: () =>
2100 observeRpcStream(
2101 WS_METHODS.pullRequestsSubscribeRefreshes,
2102 pullRequests.subscribeRefreshes,
2103 { "rpc.aggregate": "pull-requests" },
2104 ),
20992105 [WS_METHODS.pullRequestsReviewerCandidates]: (input) =>
21002106 observeRpcEffect(
21012107 WS_METHODS.pullRequestsReviewerCandidates,
apps/web/src/components/pullRequest/PullRequestDetailPanel.tsx+16-6
......@@ -64,7 +64,11 @@ import { useProjects } from "~/state/entities";
6464import { useEnvironments } from "~/state/environments";
6565import { useEnvironmentQuery } from "~/state/query";
6666import { useLiveRefresh } from "~/hooks/useLiveRefresh";
67import { pullRequestEnvironment, useSharedPullRequestSummary } from "~/state/pullRequests";
67import {
68 pullRequestEnvironment,
69 usePullRequestTurnRefresh,
70 useSharedPullRequestSummary,
71} from "~/state/pullRequests";
6872import { useAtomCommand } from "~/state/use-atom-command";
6973import { vcsEnvironment } from "~/state/vcs";
7074import { formatRelativeTimeLabel } from "~/timestampFormat";
......@@ -577,6 +581,7 @@ export function PullRequestDetailPanel({
577581 const activityQuery = useEnvironmentQuery(
578582 pullRequestEnvironment.activity({ environmentId, input: reference }),
579583 );
584 const turnRefresh = usePullRequestTurnRefresh(environmentId);
580585 const [cachedDetail, setCachedDetail] = useState(() =>
581586 readPullRequestDetailSnapshot(
582587 typeof window === "undefined" ? undefined : window.localStorage,
......@@ -675,6 +680,8 @@ export function PullRequestDetailPanel({
675680 detailQuery.refresh();
676681 activityQuery.refresh();
677682 }, [activityQuery.refresh, detailQuery.refresh]);
683 const [refreshToken, setRefreshToken] = useState(0);
684 const codeRefreshToken = refreshToken + (turnRefresh ?? 0);
678685 const activityRevision = useRef<{ readonly key: string; readonly updatedAt: string } | null>(
679686 null,
680687 );
......@@ -698,15 +705,18 @@ export function PullRequestDetailPanel({
698705 // revision effect above reads it only after this same pull request reports a change. Keyed by
699706 // the pull request rather than by the panel, because this one panel shows a different pull
700707 // request every time it is opened.
701 useLiveRefresh(detailQuery.refresh, {
702 key: `pull-request:${reference.projectId}:${reference.repository}#${reference.number}`,
703 });
708 useLiveRefresh(
709 () => {
710 detailQuery.refresh();
711 setRefreshToken((token) => token + 1);
712 },
713 { key: `pull-request:${reference.projectId}:${reference.repository}#${reference.number}` },
714 );
704715 // The button, on the other hand, goes around the server's cache rather than through it: it is
705716 // the answer for a reader who can see that what they are looking at is behind. The
706717 // invalidation goes first so the re-reads miss that cache; if it fails, the reads still run
707718 // and at worst answer from it.
708719 const invalidate = useAtomCommand(pullRequestEnvironment.invalidate, { reportFailure: false });
709 const [refreshToken, setRefreshToken] = useState(0);
710720 const refreshFromHost = useCallback(async () => {
711721 await invalidate({ environmentId, input: { reference } });
712722 refreshDetail();
......@@ -2352,7 +2362,7 @@ export function PullRequestDetailPanel({
23522362 fixFindingLabel={handoffLabels.fixFinding}
23532363 onFixFinding={startFixFinding}
23542364 onRefresh={refreshDetail}
2355 refreshToken={refreshToken}
2365 refreshToken={codeRefreshToken}
23562366 />
23572367 </Suspense>
23582368 </div>
apps/web/src/routes/_chat.pull-requests.tsx+19
......@@ -135,6 +135,7 @@ import {
135135 pullRequestEnvironment,
136136 usePullRequestList,
137137 usePullRequestListStats,
138 usePullRequestTurnRefreshes,
138139 type EnvironmentQueryTarget,
139140} from "../state/pullRequests";
140141import { useAtomCommand } from "../state/use-atom-command";
......@@ -632,6 +633,12 @@ function PullRequestsRouteView() {
632633 .join("|"),
633634 [environmentQueries],
634635 );
636 const turnRefreshes = usePullRequestTurnRefreshes(
637 environmentQueries.map(({ environmentId }) => environmentId),
638 );
639 const turnRefreshToken = turnRefreshes
640 .map(([environmentId, revision]) => `${environmentId}:${revision}`)
641 .join("|");
635642 // Page size is view state, not a URL concern: a shared link should open the first page.
636643 const scopeKey = `${environmentKey}:${assignmentKey}:${search.state}:${search.involvement}:${scopedProjectId ?? ""}:${search.host ?? ""}:${search.draft ?? ""}:${search.review ?? ""}:${search.checks ?? ""}:${search.author ?? ""}:${search.labels?.join("\u0000") ?? ""}`;
637644 const filterKey = `${scopeKey}:${sentQuery}`;
......@@ -1091,6 +1098,18 @@ function PullRequestsRouteView() {
10911098 });
10921099 };
10931100
1101 const appliedTurnRefreshToken = useRef("");
1102 const refreshAfterTurn = useEffectEvent(() => {
1103 if (sentCursors !== null) refreshList();
1104 });
1105 useEffect(() => {
1106 if (turnRefreshToken.length === 0 || appliedTurnRefreshToken.current === turnRefreshToken) {
1107 return;
1108 }
1109 appliedTurnRefreshToken.current = turnRefreshToken;
1110 refreshAfterTurn();
1111 }, [turnRefreshToken]);
1112
10941113 // The list goes stale the same way the detail does: somebody opens a pull request, a check
10951114 // finishes, a branch is merged. So it reads again on the way back to the window, and once a
10961115 // minute while somebody is reading it. Those reads go through the server's cache and stop
apps/web/src/state/pullRequests.ts+23-2
......@@ -24,8 +24,10 @@ import {
2424import { formatEnvironmentQueryError } from "./query";
2525
2626export const pullRequestEnvironment = createPullRequestEnvironmentAtoms(connectionAtomRuntime);
27export const linkedPullRequestDetailAtom =
28 createLinkedPullRequestSummaryAtomFamily(connectionAtomRuntime);
27export const linkedPullRequestDetailAtom = createLinkedPullRequestSummaryAtomFamily(
28 connectionAtomRuntime,
29 pullRequestEnvironment.refreshes,
30);
2931
3032const observedPullRequestSummaryAtom = Atom.family((key: string) =>
3133 Atom.make<PullRequestSummary | null>(null).pipe(
......@@ -148,6 +150,25 @@ const usePullRequestStatsQuery = createMergedEnvironmentQuery(
148150 pullRequestEnvironment.listStats,
149151);
150152
153const usePullRequestTurnRefreshQuery = createMergedEnvironmentQuery(
154 "web-pull-requests:turn-refreshes",
155 ({ environmentId }: EnvironmentQueryTarget<Readonly<Record<string, never>>>) =>
156 pullRequestEnvironment.refreshes({ environmentId, input: {} }),
157);
158
159export function usePullRequestTurnRefreshes(
160 environmentIds: ReadonlyArray<EnvironmentId>,
161): ReadonlyArray<readonly [EnvironmentId, number]> {
162 return usePullRequestTurnRefreshQuery(
163 environmentIds.map((environmentId) => ({ environmentId, input: {} })),
164 ).values;
165}
166
167export function usePullRequestTurnRefresh(environmentId: EnvironmentId): number | null {
168 const result = useAtomValue(pullRequestEnvironment.refreshes({ environmentId, input: {} }));
169 return Option.getOrNull(AsyncResult.value(result));
170}
171
151172export interface MergedPullRequestListView {
152173 readonly data: MergedPullRequestList | null;
153174 readonly error: string | null;
packages/client-runtime/src/rpc/client.ts+1
......@@ -52,6 +52,7 @@ export type EnvironmentSubscriptionRpcTag =
5252 | typeof WS_METHODS.subscribePreviewEvents
5353 | typeof WS_METHODS.subscribeDiscoveredLocalServers
5454 | typeof WS_METHODS.subscribeResourceTelemetry
55 | typeof WS_METHODS.pullRequestsSubscribeRefreshes
5556 | typeof WS_METHODS.previewAutomationConnect
5657 | typeof WS_METHODS.subscribeVcsStatus
5758 | typeof WS_METHODS.terminalAttach;
packages/client-runtime/src/state/pullRequests.test.ts+20
......@@ -1,8 +1,10 @@
11import { EnvironmentId, ProjectId, WS_METHODS } from "@t3tools/contracts";
22import { expect, it } from "@effect/vitest";
33import * as Effect from "effect/Effect";
4import * as Latch from "effect/Latch";
45import * as Layer from "effect/Layer";
56import * as Option from "effect/Option";
7import * as PubSub from "effect/PubSub";
68import * as Stream from "effect/Stream";
79import * as SubscriptionRef from "effect/SubscriptionRef";
810import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
......@@ -42,8 +44,10 @@ function session(client: WsRpcProtocolClient): RpcSession {
4244it.effect("refreshes pull request activity after a comment is updated", () =>
4345 Effect.scoped(
4446 Effect.gen(function* () {
47 const refreshEvents = yield* PubSub.unbounded<number>();
4548 let commentBody = "old comment";
4649 const client = {
50 [WS_METHODS.pullRequestsSubscribeRefreshes]: () => Stream.fromPubSub(refreshEvents),
4751 [WS_METHODS.pullRequestsActivity]: () =>
4852 Effect.succeed({
4953 author: null,
......@@ -138,6 +142,22 @@ it.effect("refreshes pull request activity after a comment is updated", () =>
138142 (yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).comments[0]
139143 ?.body,
140144 ).toBe("updated");
145 const refreshed = Latch.makeUnsafe();
146 const stop = registry.subscribe(activity, (result) => {
147 if (AsyncResult.isSuccess(result) && result.value.comments[0]?.body === "after turn") {
148 refreshed.openUnsafe();
149 }
150 });
151 yield* Effect.addFinalizer(() => Effect.sync(stop));
152
153 commentBody = "after turn";
154 yield* PubSub.publish(refreshEvents, 1);
155 yield* refreshed.await;
156
157 expect(
158 (yield* AtomRegistry.getResult(registry, activity, { suspendOnWaiting: true })).comments[0]
159 ?.body,
160 ).toBe("after turn");
141161 }),
142162 ),
143163);
packages/client-runtime/src/state/pullRequests.ts+19
......@@ -15,6 +15,7 @@ import {
1515 createAtomCommandScheduler,
1616 createEnvironmentRpcCommand,
1717 createEnvironmentRpcQueryAtomFamily,
18 createEnvironmentRpcSubscriptionAtomFamily,
1819 createEnvironmentQueryAtomFamily,
1920} from "./runtime.ts";
2021import { PullRequestDiffLoader } from "./pullRequestDiffHttp.ts";
......@@ -34,9 +35,19 @@ export class EnvironmentHttpConnectionNotReadyError extends Data.TaggedError(
3435
3536export const LINKED_PULL_REQUEST_IDLE_TTL_MS = 5_000;
3637
38function createPullRequestRefreshAtomFamily<R, E>(
39 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
40) {
41 return createEnvironmentRpcSubscriptionAtomFamily(runtime, {
42 label: "environment-data:pull-requests:turn-refreshes",
43 tag: WS_METHODS.pullRequestsSubscribeRefreshes,
44 });
45}
46
3747/** Refresh only the live fields a linked thread renders. */
3848export function createLinkedPullRequestSummaryAtomFamily<R, E>(
3949 runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,
50 refreshes = createPullRequestRefreshAtomFamily(runtime),
4051) {
4152 return createEnvironmentRpcQueryAtomFamily(runtime, {
4253 label: "environment-data:pull-requests:linked-summary",
......@@ -44,6 +55,7 @@ export function createLinkedPullRequestSummaryAtomFamily<R, E>(
4455 staleTimeMs: 60_000,
4556 refreshIntervalMs: 60_000,
4657 idleTtlMs: LINKED_PULL_REQUEST_IDLE_TTL_MS,
58 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
4759 });
4860}
4961
......@@ -69,6 +81,7 @@ export function pullRequestDetailToVcsStatus(
6981export function createPullRequestEnvironmentAtoms<R, E>(
7082 runtime: Atom.AtomRuntime<EnvironmentRegistry | PullRequestDiffLoader | R, E>,
7183) {
84 const refreshes = createPullRequestRefreshAtomFamily(runtime);
7285 const commandScheduler = createAtomCommandScheduler();
7386 const serialPerEnvironment = {
7487 mode: "serial",
......@@ -78,12 +91,16 @@ export function createPullRequestEnvironmentAtoms<R, E>(
7891 label: "environment-data:pull-requests:activity",
7992 tag: WS_METHODS.pullRequestsActivity,
8093 staleTimeMs: 15_000,
94 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
8195 });
8296 return {
97 refreshes,
8398 list: createEnvironmentRpcQueryAtomFamily(runtime, {
8499 label: "environment-data:pull-requests:list",
85100 tag: WS_METHODS.pullRequestsList,
86101 staleTimeMs: 30_000,
102 refreshTrigger: ({ environmentId, input }) =>
103 input.cursors === undefined ? refreshes({ environmentId, input: {} }) : undefined,
87104 }),
88105 /**
89106 * The line counts for rows the listing has already handed over. Its own query because the
......@@ -95,11 +112,13 @@ export function createPullRequestEnvironmentAtoms<R, E>(
95112 label: "environment-data:pull-requests:list-stats",
96113 tag: WS_METHODS.pullRequestsListStats,
97114 staleTimeMs: 60_000,
115 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
98116 }),
99117 detail: createEnvironmentRpcQueryAtomFamily(runtime, {
100118 label: "environment-data:pull-requests:detail",
101119 tag: WS_METHODS.pullRequestsDetail,
102120 staleTimeMs: 15_000,
121 refreshTrigger: ({ environmentId }) => refreshes({ environmentId, input: {} }),
103122 }),
104123 activity,
105124 threadComments: createEnvironmentRpcCommand(runtime, {
packages/client-runtime/src/state/runtime.ts+16-2
......@@ -51,6 +51,10 @@ interface EnvironmentQueryAtomOptions<Input, A, E, R> extends EnvironmentAtomOpt
5151 readonly staleTimeMs?: number;
5252 readonly idleTtlMs?: number;
5353 readonly refreshIntervalMs?: number;
54 readonly refreshTrigger?: (target: {
55 readonly environmentId: EnvironmentIdType;
56 readonly input: Input;
57 }) => Atom.Atom<unknown> | undefined;
5458}
5559
5660interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> {
......@@ -565,10 +569,15 @@ export function createEnvironmentQueryAtomFamily<R, ER, Input, A, E>(
565569 }),
566570 Atom.setIdleTTL(idleTtlMs),
567571 );
568 return (
572 const intervalQuery =
569573 options.refreshIntervalMs === undefined
570574 ? queryAtom
571 : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs))
575 : queryAtom.pipe(Atom.withRefresh(options.refreshIntervalMs));
576 const refreshTrigger = options.refreshTrigger?.(target);
577 return (
578 refreshTrigger === undefined
579 ? intervalQuery
580 : intervalQuery.pipe(Atom.makeRefreshOnSignal(refreshTrigger))
572581 ).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(`${options.label}:${key}`));
573582 });
574583 return (target) => family(environmentRpcKey(target));
......@@ -615,6 +624,10 @@ export function createEnvironmentRpcQueryAtomFamily<R, ER, TTag extends Environm
615624 readonly staleTimeMs?: number;
616625 readonly idleTtlMs?: number;
617626 readonly refreshIntervalMs?: number;
627 readonly refreshTrigger?: (target: {
628 readonly environmentId: EnvironmentIdType;
629 readonly input: EnvironmentRpcInput<TTag>;
630 }) => Atom.Atom<unknown> | undefined;
618631 },
619632) {
620633 return createEnvironmentQueryAtomFamily(runtime, {
......@@ -624,6 +637,7 @@ export function createEnvironmentRpcQueryAtomFamily<R, ER, TTag extends Environm
624637 ...(options.refreshIntervalMs === undefined
625638 ? {}
626639 : { refreshIntervalMs: options.refreshIntervalMs }),
640 ...(options.refreshTrigger === undefined ? {} : { refreshTrigger: options.refreshTrigger }),
627641 execute: (input: EnvironmentRpcInput<TTag>) => request(options.tag, input),
628642 });
629643}
packages/contracts/src/rpc.ts+13-1
......@@ -1,7 +1,7 @@
11import * as Schema from "effect/Schema";
22import * as Rpc from "effect/unstable/rpc/Rpc";
33import * as RpcGroup from "effect/unstable/rpc/RpcGroup";
4import { TrimmedNonEmptyString } from "./baseSchemas.ts";
4import { NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts";
55import {
66 ProviderAuthCancelInput,
77 ProviderAuthCompleteInput,
......@@ -343,6 +343,7 @@ export const WS_METHODS = {
343343 pullRequestsSetThreadResolution: "pullRequests.setThreadResolution",
344344 pullRequestsSetReaction: "pullRequests.setReaction",
345345 pullRequestsInvalidate: "pullRequests.invalidate",
346 pullRequestsSubscribeRefreshes: "pullRequests.subscribeRefreshes",
346347 pullRequestsReviewerCandidates: "pullRequests.reviewerCandidates",
347348 pullRequestsRequestReviewers: "pullRequests.requestReviewers",
348349 pullRequestsLabelCandidates: "pullRequests.labelCandidates",
......@@ -714,6 +715,16 @@ export const WsPullRequestsInvalidateRpc = Rpc.make(WS_METHODS.pullRequestsInval
714715 error: PullRequestRpcError,
715716});
716717
718export const WsPullRequestsSubscribeRefreshesRpc = Rpc.make(
719 WS_METHODS.pullRequestsSubscribeRefreshes,
720 {
721 payload: Schema.Struct({}),
722 success: NonNegativeInt,
723 error: EnvironmentAuthorizationError,
724 stream: true,
725 },
726);
727
717728/**
718729 * Read on its own rather than as part of the detail: the people who may be asked are only wanted
719730 * 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(
12151226 WsPullRequestsSetThreadResolutionRpc,
12161227 WsPullRequestsSetReactionRpc,
12171228 WsPullRequestsInvalidateRpc,
1229 WsPullRequestsSubscribeRefreshesRpc,
12181230 WsPullRequestsReviewerCandidatesRpc,
12191231 WsPullRequestsRequestReviewersRpc,
12201232 WsPullRequestsLabelCandidatesRpc,