authorgravatar for eggfriedricew.g.o@gmail.comeggfriedrice <eggfriedricew.g.o@gmail.com> 2026-03-10 23:54:47+00:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-03-10 16:54:47-07:00
loge8b0126371a0d72b2a2ad71a4829e452a109a566
tree5ed91761bbc10b4f99c61c6472b531e0379794b5
parent90d9a2ad1f1810b27d88f0dfc38304ec5b1de566
signature Signed by PGP key B5690EEEBB952194

fix: checkpoint diffs never resolve (shared PubSub subscription) (#595)

5 files changed, 297 insertions(+), 139 deletions(-)

apps/server/src/orchestration/Layers/CheckpointReactor.ts+247-118
......@@ -2,6 +2,7 @@ import {
22 CommandId,
33 EventId,
44 MessageId,
5 type ProjectId,
56 ThreadId,
67 TurnId,
78 type OrchestrationEvent,
......@@ -147,88 +148,89 @@ const make = Effect.gen(function* () {
147148
148149 const isGitWorkspace = (cwd: string) => isGitRepository(cwd);
149150
150 const captureCheckpointFromTurnCompletion = Effect.fnUntraced(function* (
151 event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>,
152 ) {
153 const turnId = toTurnId(event.turnId);
154 if (!turnId) {
155 return;
156 }
157
158 const readModel = yield* orchestrationEngine.getReadModel();
159 const thread = readModel.threads.find((entry) => entry.id === event.threadId);
160 if (!thread) {
161 return;
162 }
163
164 // When a primary turn is active, only that turn may produce completion checkpoints.
165 if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) {
166 return;
167 }
168
169 if (thread.checkpoints.some((checkpoint) => checkpoint.turnId === turnId)) {
170 return;
171 }
151 // Resolves the workspace CWD for checkpoint operations, preferring the
152 // active provider session CWD and falling back to the thread/project config.
153 // Returns undefined when no CWD can be determined or the workspace is not
154 // a git repository.
155 const resolveCheckpointCwd = Effect.fnUntraced(function* (input: {
156 readonly threadId: ThreadId;
157 readonly thread: { readonly projectId: ProjectId; readonly worktreePath: string | null };
158 readonly projects: ReadonlyArray<{ readonly id: ProjectId; readonly workspaceRoot: string }>;
159 readonly preferSessionRuntime: boolean;
160 }): Effect.fn.Return<string | undefined> {
161 const fromSession = yield* resolveSessionRuntimeForThread(input.threadId);
162 const fromThread = resolveThreadWorkspaceCwd({
163 thread: input.thread,
164 projects: input.projects,
165 });
172166
173 const sessionRuntime = yield* resolveSessionRuntimeForThread(thread.id);
174 const checkpointCwd =
175 Option.match(sessionRuntime, {
176 onNone: () => undefined,
177 onSome: (runtime) => runtime.cwd,
178 }) ??
179 resolveThreadWorkspaceCwd({
180 thread,
181 projects: readModel.projects,
182 });
183 if (!checkpointCwd) {
184 yield* Effect.logWarning("checkpoint capture skipped: no active provider session cwd", {
185 threadId: thread.id,
186 turnId,
187 });
188 return;
167 const cwd = input.preferSessionRuntime
168 ? (Option.match(fromSession, {
169 onNone: () => undefined,
170 onSome: (runtime) => runtime.cwd,
171 }) ?? fromThread)
172 : (fromThread ??
173 Option.match(fromSession, {
174 onNone: () => undefined,
175 onSome: (runtime) => runtime.cwd,
176 }));
177
178 if (!cwd) {
179 return undefined;
189180 }
190 if (!isGitWorkspace(checkpointCwd)) {
191 yield* Effect.logDebug("checkpoint capture skipped for non-git workspace", {
192 threadId: thread.id,
193 turnId,
194 cwd: checkpointCwd,
195 });
196 return;
181 if (!isGitWorkspace(cwd)) {
182 return undefined;
197183 }
184 return cwd;
185 });
198186
199 const currentTurnCount = thread.checkpoints.reduce(
200 (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),
201 0,
202 );
203 const nextTurnCount = currentTurnCount + 1;
204 const fromTurnCount = Math.max(0, nextTurnCount - 1);
205 const fromCheckpointRef = checkpointRefForThreadTurn(thread.id, fromTurnCount);
206 const targetCheckpointRef = checkpointRefForThreadTurn(thread.id, nextTurnCount);
187 // Shared tail for both capture paths: creates the git checkpoint ref, diffs
188 // it against the previous turn, then dispatches the domain events to update
189 // the orchestration read model.
190 const captureAndDispatchCheckpoint = Effect.fnUntraced(function* (input: {
191 readonly threadId: ThreadId;
192 readonly turnId: TurnId;
193 readonly thread: {
194 readonly messages: ReadonlyArray<{
195 readonly id: MessageId;
196 readonly role: string;
197 readonly turnId: TurnId | null;
198 }>;
199 };
200 readonly cwd: string;
201 readonly turnCount: number;
202 readonly status: "ready" | "missing" | "error";
203 readonly assistantMessageId: MessageId | undefined;
204 readonly createdAt: string;
205 }) {
206 const fromTurnCount = Math.max(0, input.turnCount - 1);
207 const fromCheckpointRef = checkpointRefForThreadTurn(input.threadId, fromTurnCount);
208 const targetCheckpointRef = checkpointRefForThreadTurn(input.threadId, input.turnCount);
207209
208210 const fromCheckpointExists = yield* checkpointStore.hasCheckpointRef({
209 cwd: checkpointCwd,
211 cwd: input.cwd,
210212 checkpointRef: fromCheckpointRef,
211213 });
212214 if (!fromCheckpointExists) {
213 yield* Effect.logWarning("checkpoint completion missing pre-turn baseline", {
214 threadId: thread.id,
215 turnId,
215 yield* Effect.logWarning("checkpoint capture missing pre-turn baseline", {
216 threadId: input.threadId,
217 turnId: input.turnId,
216218 fromTurnCount,
217219 });
218220 }
219221
220222 yield* checkpointStore.captureCheckpoint({
221 cwd: checkpointCwd,
223 cwd: input.cwd,
222224 checkpointRef: targetCheckpointRef,
223225 });
224226
225227 // Invalidate the workspace entry cache so the @-mention file picker
226228 // reflects files created or deleted during this turn.
227 clearWorkspaceIndexCache(checkpointCwd);
229 clearWorkspaceIndexCache(input.cwd);
228230
229231 const files = yield* checkpointStore
230232 .diffCheckpoints({
231 cwd: checkpointCwd,
233 cwd: input.cwd,
232234 fromCheckpointRef,
233235 toCheckpointRef: targetCheckpointRef,
234236 fallbackFromToHead: false,
......@@ -244,81 +246,82 @@ const make = Effect.gen(function* () {
244246 ),
245247 Effect.tapError((error) =>
246248 appendCaptureFailureActivity({
247 threadId: thread.id,
248 turnId,
249 threadId: input.threadId,
250 turnId: input.turnId,
249251 detail: `Checkpoint captured, but turn diff summary is unavailable: ${error.message}`,
250 createdAt: event.createdAt,
252 createdAt: input.createdAt,
251253 }),
252254 ),
253255 Effect.catch((error) =>
254256 Effect.logWarning("failed to derive checkpoint file summary", {
255 threadId: thread.id,
256 turnId,
257 turnCount: nextTurnCount,
257 threadId: input.threadId,
258 turnId: input.turnId,
259 turnCount: input.turnCount,
258260 detail: error.message,
259261 }).pipe(Effect.as([])),
260262 ),
261263 );
262264
263265 const assistantMessageId =
264 thread.messages
266 input.assistantMessageId ??
267 input.thread.messages
265268 .toReversed()
266 .find((entry) => entry.role === "assistant" && entry.turnId === turnId)?.id ??
267 MessageId.makeUnsafe(`assistant:${turnId}`);
269 .find((entry) => entry.role === "assistant" && entry.turnId === input.turnId)?.id ??
270 MessageId.makeUnsafe(`assistant:${input.turnId}`);
268271
269 const now = event.createdAt;
270272 yield* orchestrationEngine.dispatch({
271273 type: "thread.turn.diff.complete",
272274 commandId: serverCommandId("checkpoint-turn-diff-complete"),
273 threadId: thread.id,
274 turnId,
275 completedAt: now,
275 threadId: input.threadId,
276 turnId: input.turnId,
277 completedAt: input.createdAt,
276278 checkpointRef: targetCheckpointRef,
277 status: checkpointStatusFromRuntime(event.payload.state),
279 status: input.status,
278280 files,
279281 assistantMessageId,
280 checkpointTurnCount: nextTurnCount,
281 createdAt: now,
282 checkpointTurnCount: input.turnCount,
283 createdAt: input.createdAt,
282284 });
283285 yield* receiptBus.publish({
284286 type: "checkpoint.diff.finalized",
285 threadId: thread.id,
286 turnId,
287 checkpointTurnCount: nextTurnCount,
287 threadId: input.threadId,
288 turnId: input.turnId,
289 checkpointTurnCount: input.turnCount,
288290 checkpointRef: targetCheckpointRef,
289 status: checkpointStatusFromRuntime(event.payload.state),
290 createdAt: now,
291 status: input.status,
292 createdAt: input.createdAt,
291293 });
292294 yield* receiptBus.publish({
293295 type: "turn.processing.quiesced",
294 threadId: thread.id,
295 turnId,
296 checkpointTurnCount: nextTurnCount,
297 createdAt: now,
296 threadId: input.threadId,
297 turnId: input.turnId,
298 checkpointTurnCount: input.turnCount,
299 createdAt: input.createdAt,
298300 });
299301
300302 yield* orchestrationEngine.dispatch({
301303 type: "thread.activity.append",
302304 commandId: serverCommandId("checkpoint-captured-activity"),
303 threadId: thread.id,
305 threadId: input.threadId,
304306 activity: {
305307 id: EventId.makeUnsafe(crypto.randomUUID()),
306308 tone: "info",
307309 kind: "checkpoint.captured",
308310 summary: "Checkpoint captured",
309311 payload: {
310 turnCount: nextTurnCount,
311 status: event.payload.state,
312 turnCount: input.turnCount,
313 status: input.status,
312314 },
313 turnId,
314 createdAt: now,
315 turnId: input.turnId,
316 createdAt: input.createdAt,
315317 },
316 createdAt: now,
318 createdAt: input.createdAt,
317319 });
318320 });
319321
320 const ensurePreTurnBaselineFromTurnStart = Effect.fnUntraced(function* (
321 event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>,
322 // Captures a real git checkpoint when a turn completes via a runtime event.
323 const captureCheckpointFromTurnCompletion = Effect.fnUntraced(function* (
324 event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>,
322325 ) {
323326 const turnId = toTurnId(event.turnId);
324327 if (!turnId) {
......@@ -331,24 +334,140 @@ const make = Effect.gen(function* () {
331334 return;
332335 }
333336
334 const checkpointCwdFromThreadOrProject = resolveThreadWorkspaceCwd({
337 // When a primary turn is active, only that turn may produce completion checkpoints.
338 if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) {
339 return;
340 }
341
342 // Only skip if a real (non-placeholder) checkpoint already exists for this turn.
343 // ProviderRuntimeIngestion may insert placeholder entries with status "missing"
344 // before this reactor runs; those must not prevent real git capture.
345 if (
346 thread.checkpoints.some(
347 (checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",
348 )
349 ) {
350 return;
351 }
352
353 const checkpointCwd = yield* resolveCheckpointCwd({
354 threadId: thread.id,
335355 thread,
336356 projects: readModel.projects,
357 preferSessionRuntime: true,
337358 });
338 const checkpointCwd =
339 checkpointCwdFromThreadOrProject ??
340 Option.match(yield* resolveSessionRuntimeForThread(thread.id), {
341 onNone: () => undefined,
342 onSome: (runtime) => runtime.cwd,
343 });
344359 if (!checkpointCwd) {
345 yield* Effect.logWarning("checkpoint pre-turn capture skipped: no workspace cwd", {
346 threadId: thread.id,
347 turnId,
360 return;
361 }
362
363 // If a placeholder checkpoint exists for this turn, reuse its turn count
364 // instead of incrementing past it.
365 const existingPlaceholder = thread.checkpoints.find(
366 (checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",
367 );
368 const currentTurnCount = thread.checkpoints.reduce(
369 (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),
370 0,
371 );
372 const nextTurnCount = existingPlaceholder
373 ? existingPlaceholder.checkpointTurnCount
374 : currentTurnCount + 1;
375
376 yield* captureAndDispatchCheckpoint({
377 threadId: thread.id,
378 turnId,
379 thread,
380 cwd: checkpointCwd,
381 turnCount: nextTurnCount,
382 status: checkpointStatusFromRuntime(event.payload.state),
383 assistantMessageId: undefined,
384 createdAt: event.createdAt,
385 });
386 });
387
388 // Captures a real git checkpoint when a placeholder checkpoint (status "missing")
389 // is detected via a domain event. This replaces the placeholder with a real
390 // git-ref-based checkpoint.
391 //
392 // ProviderRuntimeIngestion creates placeholder checkpoints on turn.diff.updated
393 // events from the Codex runtime. This handler fires when the corresponding
394 // domain event arrives, allowing the reactor to capture the actual filesystem
395 // state into a git ref and dispatch a replacement checkpoint.
396 const captureCheckpointFromPlaceholder = Effect.fnUntraced(function* (
397 event: Extract<OrchestrationEvent, { type: "thread.turn-diff-completed" }>,
398 ) {
399 const { threadId, turnId, checkpointTurnCount, status } = event.payload;
400
401 // Only replace placeholders; skip events from our own real captures.
402 if (status !== "missing") {
403 return;
404 }
405
406 const readModel = yield* orchestrationEngine.getReadModel();
407 const thread = readModel.threads.find((entry) => entry.id === threadId);
408 if (!thread) {
409 yield* Effect.logWarning("checkpoint capture from placeholder skipped: thread not found", {
410 threadId,
348411 });
349412 return;
350413 }
351 if (!isGitWorkspace(checkpointCwd)) {
414
415 // If a real checkpoint already exists for this turn, skip.
416 if (
417 thread.checkpoints.some(
418 (checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",
419 )
420 ) {
421 yield* Effect.logDebug(
422 "checkpoint capture from placeholder skipped: real checkpoint already exists",
423 { threadId, turnId },
424 );
425 return;
426 }
427
428 const checkpointCwd = yield* resolveCheckpointCwd({
429 threadId,
430 thread,
431 projects: readModel.projects,
432 preferSessionRuntime: true,
433 });
434 if (!checkpointCwd) {
435 return;
436 }
437
438 yield* captureAndDispatchCheckpoint({
439 threadId,
440 turnId,
441 thread,
442 cwd: checkpointCwd,
443 turnCount: checkpointTurnCount,
444 status: "ready",
445 assistantMessageId: event.payload.assistantMessageId ?? undefined,
446 createdAt: event.payload.completedAt,
447 });
448 });
449
450 const ensurePreTurnBaselineFromTurnStart = Effect.fnUntraced(function* (
451 event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>,
452 ) {
453 const turnId = toTurnId(event.turnId);
454 if (!turnId) {
455 return;
456 }
457
458 const readModel = yield* orchestrationEngine.getReadModel();
459 const thread = readModel.threads.find((entry) => entry.id === event.threadId);
460 if (!thread) {
461 return;
462 }
463
464 const checkpointCwd = yield* resolveCheckpointCwd({
465 threadId: thread.id,
466 thread,
467 projects: readModel.projects,
468 preferSessionRuntime: false,
469 });
470 if (!checkpointCwd) {
352471 return;
353472 }
354473
......@@ -401,23 +520,13 @@ const make = Effect.gen(function* () {
401520 return;
402521 }
403522
404 const checkpointCwdFromThreadOrProject = resolveThreadWorkspaceCwd({
523 const checkpointCwd = yield* resolveCheckpointCwd({
524 threadId,
405525 thread,
406526 projects: readModel.projects,
527 preferSessionRuntime: false,
407528 });
408 const checkpointCwd =
409 checkpointCwdFromThreadOrProject ??
410 Option.match(yield* resolveSessionRuntimeForThread(threadId), {
411 onNone: () => undefined,
412 onSome: (runtime) => runtime.cwd,
413 });
414529 if (!checkpointCwd) {
415 yield* Effect.logWarning("checkpoint pre-turn capture skipped: no workspace cwd", {
416 threadId,
417 });
418 return;
419 }
420 if (!isGitWorkspace(checkpointCwd)) {
421530 return;
422531 }
423532
......@@ -592,6 +701,25 @@ const make = Effect.gen(function* () {
592701 }),
593702 ),
594703 );
704 return;
705 }
706
707 // When ProviderRuntimeIngestion creates a placeholder checkpoint (status "missing")
708 // from a turn.diff.updated runtime event, capture the real git checkpoint to
709 // replace it. The providerService.streamEvents PubSub does not reliably deliver
710 // turn.completed runtime events to this reactor (shared subscription), so
711 // reacting to the domain event is the reliable path.
712 if (event.type === "thread.turn-diff-completed") {
713 yield* captureCheckpointFromPlaceholder(event).pipe(
714 Effect.catch((error) =>
715 appendCaptureFailureActivity({
716 threadId: event.payload.threadId,
717 turnId: event.payload.turnId,
718 detail: error.message,
719 createdAt: new Date().toISOString(),
720 }).pipe(Effect.catch(() => Effect.void)),
721 ),
722 );
595723 }
596724 });
597725
......@@ -644,7 +772,8 @@ const make = Effect.gen(function* () {
644772 if (
645773 event.type !== "thread.turn-start-requested" &&
646774 event.type !== "thread.message-sent" &&
647 event.type !== "thread.checkpoint-revert-requested"
775 event.type !== "thread.checkpoint-revert-requested" &&
776 event.type !== "thread.turn-diff-completed"
648777 ) {
649778 return Effect.void;
650779 }
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+6-4
......@@ -222,14 +222,16 @@ const makeOrchestrationEngine = Effect.gen(function* () {
222222 return yield* Deferred.await(result);
223223 });
224224
225 const streamDomainEvents: OrchestrationEngineShape["streamDomainEvents"] =
226 Stream.fromPubSub(eventPubSub);
227
228225 return {
229226 getReadModel,
230227 readEvents,
231228 dispatch,
232 streamDomainEvents,
229 // Each access creates a fresh PubSub subscription so that multiple
230 // consumers (wsServer, ProviderRuntimeIngestion, CheckpointReactor, etc.)
231 // each independently receive all domain events.
232 get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
233 return Stream.fromPubSub(eventPubSub);
234 },
233235 } satisfies OrchestrationEngineShape;
234236});
235237
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+28-16
......@@ -1064,22 +1064,34 @@ const make = Effect.gen(function* () {
10641064 if (event.type === "turn.diff.updated") {
10651065 const turnId = toTurnId(event.turnId);
10661066 if (turnId && (yield* isGitRepoForThread(thread.id))) {
1067 const assistantMessageId = MessageId.makeUnsafe(
1068 `assistant:${event.itemId ?? event.turnId ?? event.eventId}`,
1069 );
1070 yield* orchestrationEngine.dispatch({
1071 type: "thread.turn.diff.complete",
1072 commandId: providerCommandId(event, "thread-turn-diff-complete"),
1073 threadId: thread.id,
1074 turnId,
1075 completedAt: now,
1076 checkpointRef: CheckpointRef.makeUnsafe(`provider-diff:${event.eventId}`),
1077 status: "missing",
1078 files: [],
1079 assistantMessageId,
1080 checkpointTurnCount: thread.checkpoints.length + 1,
1081 createdAt: now,
1082 });
1067 // Skip if a checkpoint already exists for this turn. A real
1068 // (non-placeholder) capture from CheckpointReactor should not
1069 // be clobbered, and dispatching a duplicate placeholder for the
1070 // same turnId would produce an unstable checkpointTurnCount.
1071 if (thread.checkpoints.some((c) => c.turnId === turnId)) {
1072 // Already tracked; no-op.
1073 } else {
1074 const assistantMessageId = MessageId.makeUnsafe(
1075 `assistant:${event.itemId ?? event.turnId ?? event.eventId}`,
1076 );
1077 const maxTurnCount = thread.checkpoints.reduce(
1078 (max, c) => Math.max(max, c.checkpointTurnCount),
1079 0,
1080 );
1081 yield* orchestrationEngine.dispatch({
1082 type: "thread.turn.diff.complete",
1083 commandId: providerCommandId(event, "thread-turn-diff-complete"),
1084 threadId: thread.id,
1085 turnId,
1086 completedAt: now,
1087 checkpointRef: CheckpointRef.makeUnsafe(`provider-diff:${event.eventId}`),
1088 status: "missing",
1089 files: [],
1090 assistantMessageId,
1091 checkpointTurnCount: maxTurnCount + 1,
1092 createdAt: now,
1093 });
1094 }
10831095 }
10841096 }
10851097
apps/server/src/orchestration/projector.ts+10
......@@ -500,6 +500,16 @@ export function projectEvent(
500500 "checkpoint",
501501 );
502502
503 // Do not let a placeholder (status "missing") overwrite a checkpoint
504 // that has already been captured with a real git ref (status "ready").
505 // ProviderRuntimeIngestion may fire multiple turn.diff.updated events
506 // per turn; without this guard later placeholders would clobber the
507 // real capture dispatched by CheckpointReactor.
508 const existing = thread.checkpoints.find((entry) => entry.turnId === checkpoint.turnId);
509 if (existing && existing.status !== "missing" && checkpoint.status === "missing") {
510 return nextBase;
511 }
512
503513 const checkpoints = [
504514 ...thread.checkpoints.filter((entry) => entry.turnId !== checkpoint.turnId),
505515 checkpoint,
apps/server/src/provider/Layers/ProviderService.ts+6-1
......@@ -537,7 +537,12 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
537537 listSessions,
538538 getCapabilities,
539539 rollbackConversation,
540 streamEvents: Stream.fromPubSub(runtimeEventPubSub),
540 // Each access creates a fresh PubSub subscription so that multiple
541 // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each
542 // independently receive all runtime events.
543 get streamEvents(): ProviderServiceShape["streamEvents"] {
544 return Stream.fromPubSub(runtimeEventPubSub);
545 },
541546 } satisfies ProviderServiceShape;
542547 });
543548