Commit d17f46d7631b178b6834596264f0f44f210d584b | the-knife

author	Julius Marminge <julius0216@outlook.com> 2026-09-17 16:41:06-07:00
committer	GitHub <noreply@github.com> 2026-09-17 16:41:06-07:00
log	d17f46d7631b178b6834596264f0f44f210d584b
tree	fca0e071cf1b37397b29feb5cdc5a9f05b811afd
parent	67993623afb14c017bb0b959478dc892f9c427da
signature	Signed by PGP key B5690EEEBB952194

fix(server): refresh file search outside checkpoint processing (#12308)

2 files changed, 123 insertions(+), 8 deletions(-)

apps
- server
  - src
    orchestration
    Layers
    CheckpointReactor.test.ts
    CheckpointReactor.ts

apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+98-4

...	@@ -295,6 +295,7 @@ describe("CheckpointReactor", () => {	...	@@ -295,6 +295,7 @@ describe("CheckpointReactor", () => {
295	readonly checkpointLookupFailure?: (	295	readonly checkpointLookupFailure?: (
296	cwd: string,	296	cwd: string,
297	) => VcsProcessTimeoutError \| VcsProcessSpawnError \| undefined;	297	) => VcsProcessTimeoutError \| VcsProcessSpawnError \| undefined;
		298	readonly workspaceRefresh?: (cwd: string) => Effect.Effect<void>;
298	readonly hasSession?: boolean;	299	readonly hasSession?: boolean;
299	readonly seedFilesystemCheckpoints?: boolean;	300	readonly seedFilesystemCheckpoints?: boolean;
300	readonly initializeGit?: boolean;	301	readonly initializeGit?: boolean;
...	@@ -390,10 +391,10 @@ describe("CheckpointReactor", () => {	...	@@ -390,10 +391,10 @@ describe("CheckpointReactor", () => {
390	).pipe(Layer.provide(VcsDriverRegistry.layer)),	391	).pipe(Layer.provide(VcsDriverRegistry.layer)),
391	),	392	),
392	Layer.provideMerge(	393	Layer.provideMerge(
393	WorkspaceEntries.layer.pipe(	394	(options?.workspaceRefresh
394	Layer.provide(WorkspacePaths.layer),	395	? Layer.mock(WorkspaceEntries.WorkspaceEntries)({ refresh: options.workspaceRefresh })
395	Layer.provideMerge(VcsDriverRegistry.layer),	396	: WorkspaceEntries.layer
396	),	397	).pipe(Layer.provide(WorkspacePaths.layer), Layer.provideMerge(VcsDriverRegistry.layer)),
397	),	398	),
398	Layer.provideMerge(WorkspacePaths.layer),	399	Layer.provideMerge(WorkspacePaths.layer),
399	Layer.provideMerge(VcsProcess.layer),	400	Layer.provideMerge(VcsProcess.layer),
...	@@ -664,6 +665,99 @@ describe("CheckpointReactor", () => {	...	@@ -664,6 +665,99 @@ describe("CheckpointReactor", () => {
664	}),	665	}),
665	);	666	);
666		667
		668	effectIt.effect(
		669	"finalizes checkpoints in both workspaces while entry refresh is blocked and coalesces later scans",
		670	() =>
		671	Effect.gen(function* () {
		672	const entered = yield* Deferred.make<void>();
		673	const release = yield* Deferred.make<void>();
		674	const refreshCalls: string[] = [];
		675	let secondCwd = "";
		676	const harness = yield* Effect.promise(() =>
		677	createHarness({
		678	seedFilesystemCheckpoints: false,
		679	secondThreadSharingWorktree: true,
		680	secondThreadWorktreePath: () => {
		681	secondCwd = createGitRepository();
		682	tempDirs.push(secondCwd);
		683	return secondCwd;
		684	},
		685	workspaceRefresh: (cwd) =>
		686	Effect.gen(function* () {
		687	refreshCalls.push(cwd);
		688	if (refreshCalls.length === 1) {
		689	yield* Deferred.succeed(entered, undefined);
		690	yield* Deferred.await(release);
		691	}
		692	}),
		693	}),
		694	);
		695	for (const [index, threadId] of [
		696	"thread-1",
		697	"thread-2",
		698	"thread-1",
		699	"thread-1",
		700	"thread-1",
		701	].entries()) {
		702	const turnId = asTurnId(`turn-refresh-${index}`);
		703	const id = ThreadId.make(threadId);
		704	harness.provider.emit({
		705	type: "turn.started",
		706	eventId: EventId.make(`evt-refresh-start-${index}`),
		707	provider: ProviderDriverKind.make("codex"),
		708	createdAt: "2026-01-01T00:00:00.000Z",
		709	threadId: id,
		710	turnId,
		711	});
		712	if (index < 2)
		713	expect(yield* harness.nextReceipt).toMatchObject({
		714	type: "checkpoint.baseline.captured",
		715	threadId: id,
		716	});
		717	const cwd = threadId === "thread-1" ? harness.cwd : secondCwd;
		718	NodeFS.writeFileSync(NodePath.join(cwd, "README.md"), `snapshot ${index}\n`);
		719	harness.provider.emit({
		720	type: "turn.completed",
		721	eventId: EventId.make(`evt-refresh-complete-${index}`),
		722	provider: ProviderDriverKind.make("codex"),
		723	createdAt: "2026-01-01T00:00:01.000Z",
		724	threadId: id,
		725	turnId,
		726	payload: { state: "completed" },
		727	});
		728	expect(yield* harness.nextReceipt).toMatchObject({
		729	type: "checkpoint.diff.finalized",
		730	threadId: id,
		731	turnId,
		732	});
		733	expect(yield* harness.nextReceipt).toMatchObject({
		734	type: "turn.processing.quiesced",
		735	threadId: id,
		736	turnId,
		737	});
		738	if (index === 0) yield* Deferred.await(entered);
		739	}
		740	expect(refreshCalls).toEqual([harness.cwd]);
		741	expect(
		742	gitShowFileAtRef(
		743	harness.cwd,
		744	checkpointRefForThreadTurn(ThreadId.make("thread-1"), 4),
		745	"README.md",
		746	),
		747	).toBe("snapshot 4\n");
		748	expect(
		749	gitShowFileAtRef(
		750	secondCwd,
		751	checkpointRefForThreadTurn(ThreadId.make("thread-2"), 1),
		752	"README.md",
		753	),
		754	).toBe("snapshot 1\n");
		755	yield* Deferred.succeed(release, undefined);
		756	yield* Effect.promise(harness.drain);
		757	expect(refreshCalls).toEqual([harness.cwd, secondCwd, harness.cwd]);
		758	}),
		759	);
		760
667	effectIt.effect("captures baseline and large turn summaries before completion receipts", () =>	761	effectIt.effect("captures baseline and large turn summaries before completion receipts", () =>
668	Effect.gen(function* () {	762	Effect.gen(function* () {
669	const harness = yield* Effect.promise(() =>	763	const harness = yield* Effect.promise(() =>

apps/server/src/orchestration/Layers/CheckpointReactor.ts+25-4

 @@ -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<string>();
+  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<ThreadId, TurnId>();
   const pending = new Set<ThreadId>();
 @@ -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;
 });