fix(server): a finished turn reads its pull request from GitHub once

apps/server/src/orchestration-v2/ThreadPullRequestService.test.ts

@@ -1,3 +1,5 @@
+import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeServices from "@effect/platform-node/NodeServices";
 import {
   ProjectId,
   ProviderInstanceId,
@@ -9,6 +11,7 @@ import {
   type OrchestrationV2ServerCommand,
 } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
+import * as Context from "effect/Context";
 import * as SecureRandom from "effect/crypto/SecureRandom";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
@@ -22,8 +25,24 @@ import * as Queue from "effect/Queue";
 import * as Stream from "effect/Stream";
 import * as TestClock from "effect/testing/TestClock";
 
+import * as HostProcess from "@t3tools/shared/HostProcess";
+import * as SourceControlRateLimit from "@t3tools/source-control-core/server/SourceControlRateLimit";
+import * as GitHubDriver from "@t3tools/source-control-github/server/driver";
+import * as TestSourceControlHost from "@t3tools/source-control-testing/TestSourceControlHost";
+import { HttpClient, HttpClientResponse } from "effect/http";
+
 import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
+import * as ServerConfig from "../config.ts";
 import * as GitManager from "../git/GitManager.ts";
+import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";
+import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
+import * as ServerSettings from "../serverSettings.ts";
+import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts";
+import * as TextGeneration from "../textGeneration/TextGeneration.ts";
+import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
+import * as VcsDriverRegistry from "../vcs/VcsDriverRegistry.ts";
+import * as VcsProcess from "../vcs/VcsProcess.ts";
+import * as VcsProjectConfig from "../vcs/VcsProjectConfig.ts";
 import * as ProjectStore from "./ProjectStore.ts";
 import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
 import * as PullRequestService from "../pullRequest/PullRequestService.ts";
@@ -35,6 +54,146 @@ import { EventId } from "@t3tools/contracts";
 import * as Orchestrator from "./Orchestrator.ts";
 import * as ThreadPullRequestService from "./ThreadPullRequestService.ts";
 
+const GITHUB_REMOTE = "https://github.com/pingdotgg/t3code.git";
+
+// The real GitManager and GitHub driver over a local repository whose origin names GitHub but
+// pushes to a bare remote. Only the network and the gh CLI are fakes: GitHub answers each
+// GraphQL request from memory and records it, and gh hands out a token.
+const makeGitHubBackedGitManager = (
+  branch: string,
+  pullRequest: { readonly number: number; readonly title: string; readonly url: string },
+) =>
+  Effect.gen(function* () {
+    const hostRequests: Array<string> = [];
+    const openPullRequest = {
+      ...pullRequest,
+      baseRefName: "main",
+      headRefName: branch,
+      headRefOid: "f".repeat(40),
+      state: "OPEN",
+      isDraft: false,
+      mergedAt: null,
+      closedAt: null,
+      isCrossRepository: false,
+      updatedAt: "2026-08-01T00:00:00Z",
+      headRepository: { name: "t3code", nameWithOwner: "pingdotgg/t3code" },
+      headRepositoryOwner: { __typename: "Organization", login: "pingdotgg" },
+    };
+    const gitHub = HttpClient.make((request) =>
+      Effect.sync(() => {
+        const sent: { readonly variables?: Readonly<Record<string, unknown>> } =
+          request.body._tag === "Uint8Array"
+            ? JSON.parse(new TextDecoder().decode(request.body.body))
+            : {};
+        const variables = sent.variables ?? {};
+        // A batched document aliases each lookup's variables and result as `_<n>_`.
+        const aliases = [
+          ...new Set(Object.keys(variables).flatMap((key) => /^_\d+_/.exec(key) ?? [])),
+        ];
+        const lookups = aliases.length === 0 ? [""] : aliases;
+        const heads = lookups.map((alias) => String(variables[`${alias}head`]));
+        hostRequests.push(`${request.method} ${request.url} ${heads.join(",")}`);
+        const data = Object.fromEntries(
+          lookups.map((alias, index) => [
+            `${alias}repository`,
+            { pullRequests: { nodes: heads[index] === branch ? [openPullRequest] : [] } },
+          ]),
+        );
+        return HttpClientResponse.fromWeb(
+          request,
+          new Response(JSON.stringify({ data }), {
+            headers: { "content-type": "application/json" },
+          }),
+        );
+      }),
+    );
+    const platform = Layer.mergeAll(NodeServices.layer, NodeCrypto.layer);
+    const stores = yield* Layer.build(
+      Layer.merge(ProjectionStore.layer, ProjectStore.layer).pipe(
+        Layer.provideMerge(SqlitePersistence.layerMemory),
+      ),
+    );
+    const services = yield* Layer.build(
+      Layer.mergeAll(
+        Layer.mock(TextGeneration.TextGeneration)({}),
+        Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
+        Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({}),
+        ServerSettings.ServerSettingsService.layerTest(),
+        VcsDriverRegistry.layer.pipe(Layer.provide(VcsProjectConfig.layer)),
+      ).pipe(
+        Layer.provideMerge(
+          GitVcsDriver.layer.pipe(
+            Layer.provideMerge(VcsProcess.layer),
+            Layer.provideMerge(platform),
+            Layer.provideMerge(
+              ServerConfig.layerTest(process.cwd(), { prefix: "t3-thread-pr-test-" }),
+            ),
+          ),
+        ),
+        Layer.provideMerge(platform),
+      ),
+    );
+    const vcsProcess = Context.get(services, VcsProcess.VcsProcess);
+    const sourceControlHost = TestSourceControlHost.layer({
+      process: {
+        run: (input) =>
+          input.command === "gh"
+            ? Effect.succeed(
+                TestSourceControlHost.processOutput(input.args[0] === "auth" ? "token\n" : ""),
+              )
+            : vcsProcess.run(input),
+      },
+    });
+    const driver = yield* GitHubDriver.driver.make.pipe(
+      Effect.provide(
+        Layer.mergeAll(
+          sourceControlHost,
+          SourceControlRateLimit.layer,
+          Layer.succeed(HttpClient.HttpClient, gitHub),
+          platform,
+        ),
+      ),
+      // GH_HOST, GH_REPO or a token in the test's own environment must not steer the lookup.
+      Effect.provideService(HostProcess.Environment, {}),
+    );
+    const registry = yield* SourceControlProviderRegistry.makeWithProviders([
+      {
+        kind: GitHubDriver.driver.kind,
+        provider: driver.sourceControl,
+        discovery: driver.discovery,
+      },
+    ]).pipe(Effect.provide(sourceControlHost), Effect.provideContext(services));
+    const workspaceRoot = yield* Effect.gen(function* () {
+      const fileSystem = yield* FileSystem.FileSystem;
+      const path = yield* Path.Path;
+      const vcs = yield* GitVcsDriver.GitVcsDriver;
+      const git = (cwd: string, args: ReadonlyArray<string>) =>
+        vcs.execute({ operation: "ThreadPullRequestService.test.git", cwd, args });
+      const root = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-thread-pr-" });
+      const repository = path.join(root, "project");
+      const remote = path.join(root, "remote.git");
+      yield* fileSystem.makeDirectory(repository);
+      yield* git(root, ["init", "--bare", remote]);
+      yield* git(repository, ["init", "--initial-branch=main"]);
+      yield* git(repository, ["config", "user.email", "test@example.com"]);
+      yield* git(repository, ["config", "user.name", "Test User"]);
+      yield* git(repository, ["commit", "--allow-empty", "-m", "initial"]);
+      yield* git(repository, ["checkout", "-b", branch]);
+      yield* git(repository, ["remote", "add", "origin", GITHUB_REMOTE]);
+      yield* git(repository, ["remote", "set-url", "--push", "origin", remote]);
+      yield* git(repository, ["push", "-u", "origin", branch]);
+      return repository;
+    }).pipe(Effect.provideContext(services));
+    const manager = yield* GitManager.make.pipe(
+      Effect.provideContext(
+        Context.merge(services, stores).pipe(
+          Context.add(SourceControlProviderRegistry.SourceControlProviderRegistry, registry),
+        ),
+      ),
+    );
+    return { workspaceRoot, git: manager, hostRequests };
+  });
+
 describe("ThreadPullRequestServiceV2 project guard", () => {
   it("rejects a pull-request result when the project root changes before dispatch", () => {
     const currentProject = Option.some({
@@ -165,8 +324,9 @@ describe("ThreadPullRequestServiceV2 reads", () => {
 
           // A finished turn still discovers the pull request for its branch.
           yield* Queue.offer(domainEvents, {
-            type: "checkpoint.captured",
+            type: "run.updated",
             threadId: thread.id,
+            payload: { status: "completed" },
           } as OrchestrationV2DomainEvent);
           yield* Deferred.await(turnLookup);
           yield* service.drain;
@@ -181,6 +341,147 @@ describe("ThreadPullRequestServiceV2 reads", () => {
     ),
   );
 
+  // Live clock: real git subprocesses, and the GitHub driver batches lookups over a short window.
+  it.live.each([
+    {
+      commit: "a finished turn",
+      // Checkpoint capture commits the thread-start and baseline checkpoints, the run's own
+      // checkpoint and the run's end together.
+      events: [
+        { type: "checkpoint.captured" },
+        { type: "checkpoint.captured" },
+        { type: "checkpoint.captured" },
+        { type: "run.updated", payload: { status: "completed" } },
+      ],
+    },
+    {
+      commit: "rolling back one turn",
+      // Rollback marks the turn's checkpoints stale and ends with one update per rolled-back run.
+      events: [
+        { type: "checkpoint.captured", payload: { status: "stale" } },
+        { type: "checkpoint.captured", payload: { status: "stale" } },
+        { type: "run.updated", payload: { status: "rolled_back" } },
+      ],
+    },
+  ])("reads the branch pull request from the host once for $commit", ({ events }) =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const { workspaceRoot, git, hostRequests } = yield* makeGitHubBackedGitManager(
+          "feature/turn",
+          {
+            number: 7,
+            title: "Turn work",
+            url: "https://github.com/pingdotgg/t3code/pull/7",
+          },
+        );
+        let thread: OrchestrationV2ThreadShell = {
+          ...threadShell("turn-thread"),
+          branch: "feature/turn",
+        };
+        const project: OrchestrationProjectShell = {
+          id: thread.projectId,
+          title: "Project",
+          workspaceRoot,
+          defaultModelSelection: null,
+          scripts: [],
+          createdAt: "2026-08-01T00:00:00.000Z",
+          updatedAt: "2026-08-01T00:00:00.000Z",
+        };
+        const activation = yield* Deferred.make<void>();
+        const startupLinked = yield* Deferred.make<void>();
+        const startupDrained = yield* Deferred.make<void>();
+        const commitDelivered = yield* Deferred.make<void>();
+        const dependencies = Layer.mergeAll(
+          Layer.mock(Orchestrator.OrchestratorV2)({
+            streamDomainEvents: Stream.unwrap(
+              Deferred.await(startupDrained).pipe(
+                Effect.as(
+                  Stream.fromIterable(
+                    events.map(
+                      (event) => ({ ...event, threadId: thread.id }) as OrchestrationV2DomainEvent,
+                    ),
+                  ),
+                ),
+              ),
+            ).pipe(
+              Stream.concat(
+                Stream.unwrap(
+                  Deferred.succeed(commitDelivered, undefined).pipe(Effect.as(Stream.never)),
+                ),
+              ),
+            ),
+            getPullRequestSnapshot: () =>
+              Effect.sync(() => ({
+                schemaVersion: 2,
+                snapshotSequence: 1,
+                threads: [thread],
+                archivedThreads: [],
+              })),
+            dispatch: (command) =>
+              Effect.gen(function* () {
+                if (command.type === "thread.pull-request.sync") {
+                  thread = { ...thread, branchPullRequest: command.branchPullRequest };
+                  yield* Deferred.succeed(startupLinked, undefined);
+                }
+                return { sequence: 2, storedEvents: [] };
+              }),
+          }),
+          Layer.mock(ProjectStore.ProjectStoreV2)({
+            listShells: () => Effect.succeed([project]),
+            getShell: () => Effect.succeedSome(project),
+          }),
+          Layer.succeed(GitManager.GitManager, git),
+          Layer.mock(PullRequestService.PullRequestService)({}),
+          Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({
+            resolve: () =>
+              Effect.succeed({
+                canonicalKey: "github.com/pingdotgg/t3code",
+                locator: {
+                  source: "git-remote" as const,
+                  remoteName: "origin",
+                  remoteUrl: GITHUB_REMOTE,
+                },
+                provider: SourceControlProviderKind.make("github"),
+                owner: "pingdotgg",
+                name: "t3code",
+              }),
+          }),
+          Layer.mock(BackgroundPolicy.BackgroundPolicy)({
+            shouldRunScopeWork: () => Effect.succeed(false),
+            shouldRunOpportunisticWork: Effect.succeed(false),
+          }),
+          Layer.succeed(ServerActivation.ServerActivation, Deferred.await(activation)),
+          Layer.succeed(
+            SecureRandom.SecureRandom,
+            SecureRandom.make((size) => new Uint8Array(size).fill(1)),
+          ),
+          FileSystem.layerNoop({}),
+          Path.layer,
+        );
+        yield* Effect.gen(function* () {
+          const service = yield* ThreadPullRequestService.make;
+          yield* service.start();
+          yield* Deferred.succeed(activation, undefined);
+          // The startup sweep finds the pull request with one cold lookup and links it.
+          yield* Deferred.await(startupLinked);
+          yield* service.drain;
+          const oneLookup = hostRequests.splice(0);
+          expect(oneLookup.length).toBeGreaterThan(0);
+
+          yield* Deferred.succeed(startupDrained, undefined);
+          yield* Deferred.await(commitDelivered);
+          yield* service.drain;
+
+          expect(hostRequests).toEqual(oneLookup);
+          expect(thread.branchPullRequest).toMatchObject({
+            number: 7,
+            url: "https://github.com/pingdotgg/t3code/pull/7",
+          });
+        }).pipe(Effect.provide(dependencies));
+      }),
+    ),
+  );
+
   it.effect.each([
     {
       worktree: "a linked worktree",
@@ -408,8 +709,9 @@ describe("ThreadPullRequestServiceV2 reads", () => {
 
           // The child's own turn still discovers its pull request.
           yield* Queue.offer(domainEvents, {
-            type: "checkpoint.captured",
+            type: "run.updated",
             threadId: child.id,
+            payload: { status: "completed" },
           } as OrchestrationV2DomainEvent);
           yield* Deferred.await(childLookup);
           yield* service.drain;

apps/server/src/orchestration-v2/ThreadPullRequestService.ts

@@ -27,6 +27,7 @@ import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolv
 import { forkParked } from "../serverActivation.ts";
 import { sourceControlDefinitions } from "../sourceControl/sourceControlDefinitions.ts";
 import * as Orchestrator from "./Orchestrator.ts";
+import { isTerminalRunStatus } from "./ThreadManagementService.ts";
 
 class ThreadPullRequestServiceV2 extends Context.Service<
   ThreadPullRequestServiceV2,
@@ -366,15 +367,11 @@ export const make = Effect.gen(function* () {
       case "thread.metadata-updated":
         return worker.enqueue({ threadId: event.threadId, refresh: false });
       case "thread.unsettled":
-      case "checkpoint.captured":
         return worker.enqueue({ threadId: event.threadId, refresh: true });
+      // Checkpoint capture and rollback commit their checkpoints with this run update, so it
+      // alone reads the host once per finished or rolled-back run.
       case "run.updated":
-        if (
-          event.payload.status === "completed" ||
-          event.payload.status === "failed" ||
-          event.payload.status === "cancelled" ||
-          event.payload.status === "interrupted"
-        ) {
+        if (isTerminalRunStatus(event.payload.status)) {
           return worker.enqueue({ threadId: event.threadId, refresh: true });
         }
         break;