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

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"),

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,

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);
@@ -295,6 +296,7 @@ describe("CheckpointReactor", () => {
     readonly providerSessionCwd?: string;
     readonly providerName?: ProviderDriverKind;
     readonly gitStatusRefreshCalls?: Array<string>;
+    readonly pullRequestCacheRefreshCalls?: Array<number>;
     readonly pullRequestRefreshCalls?: Array<string>;
   }) {
     const cwd = createGitRepository();
@@ -355,6 +357,13 @@ describe("CheckpointReactor", () => {
       Layer.provideMerge(projectionSnapshotLayer),
       Layer.provideMerge(RuntimeReceiptBusLive),
       Layer.provideMerge(Layer.succeed(ProviderService, provider.service)),
+      Layer.provideMerge(
+        Layer.mock(PullRequestService)({
+          refreshAfterTurn: Effect.sync(() => {
+            options?.pullRequestCacheRefreshCalls?.push(1);
+          }),
+        }),
+      ),
       Layer.provideMerge(vcsStatusBroadcasterLayer),
       Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))),
       Layer.provideMerge(
@@ -545,6 +554,95 @@ describe("CheckpointReactor", () => {
     ).toBe("v2\n");
   });
 
+  effectIt.effect("refreshes once for terminal sessions, including after reactor startup", () =>
+    Effect.gen(function* () {
+      const pullRequestRefreshCalls: number[] = [];
+      const harness = yield* Effect.promise(() =>
+        createHarness({
+          seedFilesystemCheckpoints: false,
+          pullRequestCacheRefreshCalls: pullRequestRefreshCalls,
+        }),
+      );
+      const threadId = ThreadId.make("thread-1");
+      const setSession = (
+        command: string,
+        status: "running" | "ready",
+        activeTurnId: TurnId | null,
+      ) =>
+        harness.engine.dispatch({
+          type: "thread.session.set",
+          commandId: CommandId.make(`cmd-pr-refresh-${command}`),
+          threadId,
+          session: {
+            threadId,
+            status,
+            providerName: "codex",
+            runtimeMode: "approval-required",
+            activeTurnId,
+            lastError: null,
+            updatedAt: `2026-01-01T00:00:0${command === "running" ? 1 : 2}.000Z`,
+          },
+          createdAt: `2026-01-01T00:00:0${command === "running" ? 1 : 2}.000Z`,
+        });
+
+      yield* setSession("initial-ready", "ready", null);
+      yield* Effect.promise(() => harness.drain());
+      expect(pullRequestRefreshCalls).toEqual([1]);
+
+      yield* setSession("running", "running", asTurnId("turn-refresh-prs"));
+      yield* setSession("superseding", "running", asTurnId("turn-refresh-prs-2"));
+      yield* harness.engine.dispatch({
+        type: "thread.turn.diff.complete",
+        commandId: CommandId.make("cmd-pr-refresh-placeholder"),
+        threadId,
+        turnId: asTurnId("turn-refresh-prs"),
+        completedAt: "2026-01-01T00:00:01.000Z",
+        checkpointRef: checkpointRefForThreadTurn(threadId, 1),
+        checkpointTurnCount: 1,
+        status: "missing",
+        files: [],
+        createdAt: "2026-01-01T00:00:01.000Z",
+      });
+      yield* Effect.promise(() => harness.drain());
+      expect(pullRequestRefreshCalls).toEqual([1]);
+
+      yield* setSession("ready", "ready", null);
+      yield* setSession("ready-again", "ready", null);
+      yield* Effect.promise(() => harness.drain());
+
+      expect(pullRequestRefreshCalls).toEqual([1, 1]);
+    }),
+  );
+
+  effectIt.effect("refreshes pull request data from a completion without session metadata", () =>
+    Effect.gen(function* () {
+      const pullRequestRefreshCalls: number[] = [];
+      const harness = yield* Effect.promise(() =>
+        createHarness({
+          seedFilesystemCheckpoints: false,
+          pullRequestCacheRefreshCalls: pullRequestRefreshCalls,
+        }),
+      );
+      const threadId = ThreadId.make("thread-1");
+
+      yield* harness.engine.dispatch({
+        type: "thread.turn.diff.complete",
+        commandId: CommandId.make("cmd-pr-refresh-sessionless"),
+        threadId,
+        turnId: asTurnId("turn-sessionless"),
+        completedAt: "2026-01-01T00:00:01.000Z",
+        checkpointRef: checkpointRefForThreadTurn(threadId, 2),
+        checkpointTurnCount: 2,
+        status: "ready",
+        files: [],
+        createdAt: "2026-01-01T00:00:01.000Z",
+      });
+      yield* Effect.promise(() => harness.drain());
+
+      expect(pullRequestRefreshCalls).toEqual([1]);
+    }),
+  );
+
   it("refreshes local git status state on turn completion using the session cwd", async () => {
     const gitStatusRefreshCalls: string[] = [];
     const harness = await createHarness({

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,16 @@ const make = Effect.gen(function* () {
   const receiptBus = yield* RuntimeReceiptBus;
   const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
   const vcsStatusBroadcaster = yield* VcsStatusBroadcaster;
+  const pullRequests = yield* PullRequestService.PullRequestService;
+  let refreshedTurnThread: ThreadId | null = null;
+
+  const refreshPullRequestsOnce = Effect.fn("refreshPullRequestsOnce")(function* (
+    threadId: ThreadId,
+  ) {
+    if (refreshedTurnThread === threadId) return;
+    refreshedTurnThread = threadId;
+    yield* pullRequests.refreshAfterTurn;
+  });
 
   const appendRevertFailureActivity = (input: {
     readonly threadId: ThreadId;
@@ -853,7 +864,24 @@ const make = Effect.gen(function* () {
   });
 
   const processDomainEvent = Effect.fn("processDomainEvent")(function* (event: OrchestrationEvent) {
+    if (event.type === "thread.session-set") {
+      if (
+        event.payload.session.status === "starting" ||
+        event.payload.session.status === "running"
+      ) {
+        if (event.payload.session.activeTurnId !== null) {
+          refreshedTurnThread = null;
+        }
+        return;
+      }
+      yield* refreshPullRequestsOnce(event.payload.threadId);
+      return;
+    }
+
     if (event.type === "thread.turn-start-requested" || event.type === "thread.message-sent") {
+      if (event.type === "thread.turn-start-requested") {
+        refreshedTurnThread = null;
+      }
       yield* ensurePreTurnBaselineFromDomainTurnStart(event);
       return;
     }
@@ -891,13 +919,20 @@ const make = Effect.gen(function* () {
           ),
         ),
       );
+      if (event.payload.status === "missing") return;
+      const thread = yield* resolveThreadDetail(event.payload.threadId);
+      if (thread?.session === null) {
+        yield* refreshPullRequestsOnce(event.payload.threadId);
+      }
+      return;
     }
   });
 
   const processRuntimeEvent = Effect.fn("processRuntimeEvent")(function* (
     event: ProviderRuntimeEvent,
   ) {
     if (event.type === "turn.started") {
+      refreshedTurnThread = null;
       yield* ensurePreTurnBaselineFromTurnStart(event);
       return;
     }
@@ -952,6 +987,7 @@ const make = Effect.gen(function* () {
         if (
           event.type !== "thread.turn-start-requested" &&
           event.type !== "thread.message-sent" &&
+          event.type !== "thread.session-set" &&
           event.type !== "thread.checkpoint-revert-requested" &&
           event.type !== "thread.turn-diff-completed"
         ) {

apps/server/src/pullRequest/PullRequestService.test.ts

@@ -2681,6 +2681,53 @@ it.effect("an explicit invalidation makes the next listing ask the host again",
   }),
 );
 
+it.effect("refreshes listing and change request caches after a thread turn", () =>
+  Effect.scoped(
+    Effect.gen(function* () {
+      const calls: string[] = [];
+      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", {
+            listChangeRequests: (input) => {
+              calls.push(`list:${input.repository}`);
+              return Effect.succeed({
+                items: [changeRequest(1, "2026-07-02T00:00:00Z")],
+                truncated: false,
+                continues: false,
+              });
+            },
+            getChangeRequest: (input) => {
+              calls.push(`detail:${input.repository}`);
+              return Effect.succeed(hostedChangeRequest("body"));
+            },
+          }),
+        ],
+      });
+      const observedRefresh = yield* Stream.runHead(service.subscribeRefreshes).pipe(
+        Effect.forkChild({ startImmediately: true }),
+      );
+
+      yield* service.list({ state: "open", projectId: reference.projectId });
+      yield* service.detail(reference);
+      yield* service.refreshAfterTurn;
+      const refresh = Option.getOrThrow(yield* Fiber.join(observedRefresh));
+      yield* service.list({ state: "open", projectId: reference.projectId });
+      yield* service.detail(reference);
+      assert.isAbove(refresh.revision, 0);
+      assert.deepStrictEqual(calls, [
+        "list:acme/web",
+        "detail:acme/web",
+        "list:acme/web",
+        "detail:acme/web",
+      ]);
+    }),
+  ),
+);
+
 it.effect("a mutation makes the next listing ask the host again, with no client asking", () =>
   Effect.gen(function* () {
     let hostCalls = 0;

apps/server/src/pullRequest/PullRequestService.ts

@@ -39,6 +39,7 @@ import {
   type PullRequestProviderSummary,
   type PullRequestReactionInput,
   type PullRequestRef,
+  type PullRequestRefreshEvent,
   type PullRequestReviewVerdict,
   type PullRequestReviewerCandidateList,
   type PullRequestReviewerRequestInput,
@@ -148,6 +149,8 @@ export class PullRequestService extends Context.Service<
       never,
       Scope.Scope
     >;
+    readonly subscribeRefreshes: Stream.Stream<PullRequestRefreshEvent>;
+    readonly refreshAfterTurn: Effect.Effect<void>;
     readonly detail: (input: PullRequestRef) => Effect.Effect<PullRequestDetail, PullRequestError>;
     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<PullRequestMergeEvent>(64);
+  const pullRequestRefreshes = yield* PubSub.sliding<PullRequestRefreshEvent>(1);
   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<string, number>();
   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) => {
@@ -2186,6 +2192,7 @@ export const make = Effect.gen(function* () {
       // The parse undoes this module's own serialization, so the shapes are known exactly;
       // the cast restores the branded field types JSON cannot carry.
       const [
+        ,
         ,
         state,
         involvement,
@@ -2197,6 +2204,7 @@ export const make = Effect.gen(function* () {
         query,
         cursorEntries,
       ] = JSON.parse(key) as [
+        number,
         number,
         string,
         string | null,
@@ -2228,6 +2236,7 @@ export const make = Effect.gen(function* () {
   const list: PullRequestService["Service"]["list"] = (input) => {
     const key = JSON.stringify([
       listingsEpoch,
+      turnRefreshEpoch,
       input.state,
       input.involvement ?? null,
       // Positional so two identical filter sets key alike however their record was assembled.
@@ -2358,7 +2367,11 @@ export const make = Effect.gen(function* () {
 
   const listStatsCache = yield* Cache.makeWith(
     (key: string) => {
-      const [, refs] = JSON.parse(key) as [number, ReadonlyArray<[string, string, number]>];
+      const [, , refs] = JSON.parse(key) as [
+        number,
+        number,
+        ReadonlyArray<[string, string, number]>,
+      ];
       return listStatsUncached({
         refs: refs.map(([projectId, repository, number]) => ({ projectId, repository, number })),
       } as unknown as PullRequestListStatsInput);
@@ -2376,6 +2389,7 @@ export const make = Effect.gen(function* () {
     if (input.refs.length === 0) return Effect.succeed({ stats: [] });
     const key = JSON.stringify([
       listingsEpoch,
+      turnRefreshEpoch,
       input.refs
         .map((ref) => [ref.projectId, ref.repository, ref.number] as const)
         .toSorted((left, right) =>
@@ -2396,6 +2410,13 @@ export const make = Effect.gen(function* () {
     }).pipe(Effect.andThen(Cache.invalidateAll(viewerFlights)));
   };
 
+  const refreshAfterTurn: PullRequestService["Service"]["refreshAfterTurn"] = Effect.gen(
+    function* () {
+      turnRefreshEpoch = ++epochCounter;
+      yield* PubSub.publish(pullRequestRefreshes, { revision: turnRefreshEpoch });
+    },
+  ).pipe(Effect.withSpan("PullRequestService.refreshAfterTurn"));
+
   // 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 +2456,8 @@ export const make = Effect.gen(function* () {
     subscribeMerges: PubSub.subscribe(mergedPullRequests).pipe(
       Effect.map((subscription) => Stream.fromSubscription(subscription)),
     ),
+    subscribeRefreshes: Stream.fromPubSub(pullRequestRefreshes),
+    refreshAfterTurn,
     detail,
     activity,
     threadComments,