History

fix(server): capture checkpoints before refreshing PR status (#10347)

Theo Browne committed 2 months ago

Commit e0adcc81 parent 62f568b2 files · +64 −3Browse files

Unified · Split

2 files
  1. apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +45 −1
  2. apps/server/src/orchestration/Layers/CheckpointReactor.ts +19 −2
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+45−1
@@ -302,6 +302,7 @@ describe("CheckpointReactor", () => { · 301 unmodified lines
302302 readonly providerName?: ProviderDriverKind;
303303 readonly gitStatusRefreshCalls?: Array<string>;
304304 readonly pullRequestRefreshCalls?: Array<string>;
305+ readonly pullRequestRefresh?: Effect.Effect<void>;
305306 }) {
306307 const cwd = createGitRepository();
307308 if (options?.initializeGit === false) {
@@ -357,7 +358,7 @@ describe("CheckpointReactor", () => { · 49 unmodified lines
357358 refreshPullRequestStatus: (cwd: string) =>
358359 Effect.sync(() => {
359360 options?.pullRequestRefreshCalls?.push(cwd);
360− }).pipe(Effect.as(null)),
361+ }).pipe(Effect.andThen(options?.pullRequestRefresh ?? Effect.void), Effect.as(null)),
361362 streamStatus: () => Stream.empty,
362363 });
363364
@@ -879,6 +880,49 @@ describe("CheckpointReactor", () => { · 515 unmodified lines
879880 expect(pullRequestRefreshCalls).toEqual([harness.cwd]);
880881 });
881882
883+ effectIt.effect("captures files while the pull request lookup is still pending", () =>
884+ Effect.gen(function* () {
885+ const lookupStarted = yield* Deferred.make<void>();
886+ const finishLookup = yield* Deferred.make<void>();
887+ const harness = yield* Effect.promise(() =>
888+ createHarness({
889+ seedFilesystemCheckpoints: false,
890+ threadBranch: "t3code/feature",
891+ localStatusRefName: "t3code/feature",
892+ pullRequestRefresh: Deferred.succeed(lookupStarted, undefined).pipe(
893+ Effect.andThen(Deferred.await(finishLookup)),
894+ ),
895+ }),
896+ );
897+ NodeFS.writeFileSync(NodePath.join(harness.cwd, "README.md"), "completed turn\n");
898+ harness.provider.emit({
899+ type: "turn.completed",
900+ eventId: EventId.make("evt-turn-completed-slow-pr"),
901+ provider: ProviderDriverKind.make("codex"),
902+ createdAt: "2026-01-01T00:00:00.000Z",
903+ threadId: ThreadId.make("thread-1"),
904+ turnId: asTurnId("turn-slow-pr"),
905+ payload: { state: "completed" },
906+ });
907+
908+ yield* Deferred.await(lookupStarted);
909+ yield* Effect.gen(function* () {
910+ expect(yield* harness.nextReceipt).toMatchObject({
911+ type: "checkpoint.diff.finalized",
912+ turnId: "turn-slow-pr",
913+ });
914+ expect(
915+ gitShowFileAtRef(
916+ harness.cwd,
917+ checkpointRefForThreadTurn(ThreadId.make("thread-1"), 1),
918+ "README.md",
919+ ),
920+ ).toBe("completed turn\n");
921+ }).pipe(Effect.ensuring(Deferred.succeed(finishLookup, undefined)));
922+ yield* Effect.promise(harness.drain);
923+ }),
924+ );
925+
882926 it("re-asks for the pull request after adopting a drifted checkout", async () => {
883927 const pullRequestRefreshCalls: string[] = [];
884928 const harness = await createHarness({
apps/server/src/orchestration/Layers/CheckpointReactor.ts+19−2
@@ -605,6 +605,23 @@ const make = Effect.gen(function* () { · 604 unmodified lines
605605 );
606606 });
607607
608+ // Refreshing git status ends in a remote PR lookup under the vcs status
609+ // write lock. Run it on its own worker so file capture for this turn (and
610+ // checkpoints for other threads) never wait behind that network call.
611+ const statusRefreshWorker = yield* makeDrainableWorker(
612+ (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>) =>
613+ refreshLocalGitStatusFromTurnCompletion(event).pipe(
614+ Effect.catchCause((cause) =>
615+ Cause.hasInterruptsOnly(cause)
616+ ? Effect.failCause(cause)
617+ : Effect.logWarning("failed to refresh git status after turn completion", {
618+ threadId: event.threadId,
619+ cause: Cause.pretty(cause),
620+ }),
621+ ),
622+ ),
623+ );
624+
608625 const ensurePreTurnBaselineFromDomainTurnStart = Effect.fn(
609626 "ensurePreTurnBaselineFromDomainTurnStart",
610627 )(function* (
@@ -853,7 +870,7 @@ const make = Effect.gen(function* () { · 242 unmodified lines
853870 const isTrackedTurn = sameId(startedTurnId, turnId);
854871 if (isTrackedTurn) startedTurns.delete(event.threadId);
855872 if (event.type === "turn.completed") {
856− yield* refreshLocalGitStatusFromTurnCompletion(event);
873+ yield* statusRefreshWorker.enqueue(event);
857874 }
858875 if (
859876 turnId !== null &&
@@ -944,7 +961,7 @@ const make = Effect.gen(function* () { · 84 unmodified lines
944961
945962 return {
946963 start,
947− drain: worker.drain,
964+ drain: worker.drain.pipe(Effect.andThen(statusRefreshWorker.drain)),
948965 } satisfies CheckpointReactorShape;
949966 });
950967