From d17f46d7631b178b6834596264f0f44f210d584b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 17 Sep 2026 16:41:06 -0700 Subject: [PATCH] fix(server): refresh file search outside checkpoint processing (#12308) --- .../Layers/CheckpointReactor.test.ts | 102 +++++++++++++++++- .../orchestration/Layers/CheckpointReactor.ts | 29 ++++- 2 files changed, 123 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 96902a488e9e11b968c9ca0f2abc2f54367149e7..1d0b3da1bdbc9843d9e0abdc3891b939cbaece5f 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -295,6 +295,7 @@ describe("CheckpointReactor", () => { readonly checkpointLookupFailure?: ( cwd: string, ) => VcsProcessTimeoutError | VcsProcessSpawnError | undefined; + readonly workspaceRefresh?: (cwd: string) => Effect.Effect; readonly hasSession?: boolean; readonly seedFilesystemCheckpoints?: boolean; readonly initializeGit?: boolean; @@ -390,10 +391,10 @@ describe("CheckpointReactor", () => { ).pipe(Layer.provide(VcsDriverRegistry.layer)), ), Layer.provideMerge( - WorkspaceEntries.layer.pipe( - Layer.provide(WorkspacePaths.layer), - Layer.provideMerge(VcsDriverRegistry.layer), - ), + (options?.workspaceRefresh + ? Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: options.workspaceRefresh }) + : WorkspaceEntries.layer + ).pipe(Layer.provide(WorkspacePaths.layer), Layer.provideMerge(VcsDriverRegistry.layer)), ), Layer.provideMerge(WorkspacePaths.layer), Layer.provideMerge(VcsProcess.layer), @@ -664,6 +665,99 @@ describe("CheckpointReactor", () => { }), ); + effectIt.effect( + "finalizes checkpoints in both workspaces while entry refresh is blocked and coalesces later scans", + () => + Effect.gen(function* () { + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const refreshCalls: string[] = []; + let secondCwd = ""; + const harness = yield* Effect.promise(() => + createHarness({ + seedFilesystemCheckpoints: false, + secondThreadSharingWorktree: true, + secondThreadWorktreePath: () => { + secondCwd = createGitRepository(); + tempDirs.push(secondCwd); + return secondCwd; + }, + workspaceRefresh: (cwd) => + Effect.gen(function* () { + refreshCalls.push(cwd); + if (refreshCalls.length === 1) { + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + } + }), + }), + ); + for (const [index, threadId] of [ + "thread-1", + "thread-2", + "thread-1", + "thread-1", + "thread-1", + ].entries()) { + const turnId = asTurnId(`turn-refresh-${index}`); + const id = ThreadId.make(threadId); + harness.provider.emit({ + type: "turn.started", + eventId: EventId.make(`evt-refresh-start-${index}`), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:00.000Z", + threadId: id, + turnId, + }); + if (index < 2) + expect(yield* harness.nextReceipt).toMatchObject({ + type: "checkpoint.baseline.captured", + threadId: id, + }); + const cwd = threadId === "thread-1" ? harness.cwd : secondCwd; + NodeFS.writeFileSync(NodePath.join(cwd, "README.md"), `snapshot ${index}\n`); + harness.provider.emit({ + type: "turn.completed", + eventId: EventId.make(`evt-refresh-complete-${index}`), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId: id, + turnId, + payload: { state: "completed" }, + }); + expect(yield* harness.nextReceipt).toMatchObject({ + type: "checkpoint.diff.finalized", + threadId: id, + turnId, + }); + expect(yield* harness.nextReceipt).toMatchObject({ + type: "turn.processing.quiesced", + threadId: id, + turnId, + }); + if (index === 0) yield* Deferred.await(entered); + } + expect(refreshCalls).toEqual([harness.cwd]); + expect( + gitShowFileAtRef( + harness.cwd, + checkpointRefForThreadTurn(ThreadId.make("thread-1"), 4), + "README.md", + ), + ).toBe("snapshot 4\n"); + expect( + gitShowFileAtRef( + secondCwd, + checkpointRefForThreadTurn(ThreadId.make("thread-2"), 1), + "README.md", + ), + ).toBe("snapshot 1\n"); + yield* Deferred.succeed(release, undefined); + yield* Effect.promise(harness.drain); + expect(refreshCalls).toEqual([harness.cwd, secondCwd, harness.cwd]); + }), + ); + effectIt.effect("captures baseline and large turn summaries before completion receipts", () => Effect.gen(function* () { const harness = yield* Effect.promise(() => diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 7df20e6bcf36bb7a1e0f79b58306e1e13a523719..4cfdccfd8f7425afc13fde87c857c93db5c47bb6 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -93,6 +93,25 @@ const make = Effect.gen(function* () { const path = yield* Path.Path; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; const pullRequests = yield* PullRequestService.PullRequestService; + const queuedEntryRefreshes = new Set(); + const entryRefreshWorker = yield* makeDrainableWorker((cwd: string) => + Effect.sync(() => queuedEntryRefreshes.delete(cwd)).pipe( + Effect.andThen(workspaceEntries.refresh(cwd)), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("failed to refresh checkpoint workspace entries", { + cwd, + }), + ), + ), + ); + const refreshWorkspaceEntries = Effect.fn("refreshWorkspaceEntries")(function* (cwd: string) { + if (queuedEntryRefreshes.has(cwd)) return; + queuedEntryRefreshes.add(cwd); + yield* entryRefreshWorker.enqueue(cwd); + }); + const startedTurns = new Map(); const pending = new Set(); @@ -270,7 +289,7 @@ const make = Effect.gen(function* () { // Refresh the workspace entry index so the @-mention file picker // reflects files created or deleted during this turn. - yield* workspaceEntries.refresh(input.cwd); + yield* refreshWorkspaceEntries(input.cwd); // Git may have been initialized during this turn, leaving no pre-turn // snapshot. Keep the completion checkpoint for future turns, but do not @@ -629,7 +648,6 @@ const make = Effect.gen(function* () { ? Effect.failCause(cause) : Effect.logWarning("failed to refresh git status after turn completion", { threadId: event.threadId, - cause: Cause.pretty(cause), }), ), ), @@ -850,7 +868,7 @@ const make = Effect.gen(function* () { // Refresh the workspace entry index so the @-mention file picker // reflects the reverted filesystem state. - yield* workspaceEntries.refresh(checkpointCwd); + yield* refreshWorkspaceEntries(checkpointCwd); } const rolledBackTurns = Math.max(0, currentTurnCount - event.payload.turnCount); @@ -1041,7 +1059,10 @@ const make = Effect.gen(function* () { return { start, - drain: worker.drain.pipe(Effect.andThen(statusRefreshWorker.drain)), + drain: worker.drain.pipe( + Effect.andThen(statusRefreshWorker.drain), + Effect.andThen(entryRefreshWorker.drain), + ), } satisfies CheckpointReactorShape; }); -- 2.54.0