From dcf9ec417154efe6e5b13fb538095120778b923a Mon Sep 17 00:00:00 2001 From: Patrik Votocek Date: Tue, 15 Sep 2026 21:36:41 +0200 Subject: [PATCH 1/4] fix(server): keep VCS waits from blocking turn completion Provider diff events detect the Git repository through VCS subprocesses. When git was slow or hung, that single event stalled the serial ingestion worker and turn.completed sat behind it, so the turn never settled. Route turn.diff.updated to its own drainable worker so diff bookkeeping cannot delay lifecycle and message events. Co-Authored-By: Claude Fable 5.1 --- .../Layers/ProviderRuntimeIngestion.test.ts | 86 ++++++++++++++++++- .../Layers/ProviderRuntimeIngestion.ts | 12 ++- 2 files changed, 95 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 062a91f8cb53..3adcd79e17ed 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -29,6 +29,7 @@ import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as ManagedRuntime from "effect/ManagedRuntime"; import * as Option from "effect/Option"; @@ -267,6 +268,7 @@ describe("ProviderRuntimeIngestion", () => { serverSettings?: Partial; threadTitle?: string; workspaceSubdirectory?: string; + isGitRepository?: CheckpointStore.CheckpointStore["Service"]["isGitRepository"]; }) { const repositoryRoot = makeTempDir("t3-provider-project-"); NodeChildProcess.execFileSync("git", ["init", "--initial-branch=main"], { @@ -327,7 +329,15 @@ describe("ProviderRuntimeIngestion", () => { Layer.provideMerge(SqlitePersistenceMemory), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(makeTestServerSettingsLayer(options?.serverSettings)), - Layer.provideMerge(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer))), + Layer.provideMerge( + Layer.effect( + CheckpointStore.CheckpointStore, + Effect.map(CheckpointStore.CheckpointStore, (store) => ({ + ...store, + isGitRepository: options?.isGitRepository ?? store.isGitRepository, + })), + ).pipe(Layer.provide(CheckpointStore.layer.pipe(Layer.provide(VcsDriverRegistry.layer)))), + ), Layer.provideMerge(VcsProcess.layer), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), Layer.provideMerge(NodeServices.layer), @@ -3749,6 +3759,80 @@ describe("ProviderRuntimeIngestion", () => { }); }); + effectIt.effect("settles the turn while repository detection for a diff is blocked", () => + Effect.gen(function* () { + const detectionStarted = yield* Deferred.make(); + const releaseDetection = yield* Deferred.make(); + const harness = yield* Effect.promise(() => + createHarness({ + isGitRepository: () => + Deferred.succeed(detectionStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseDetection)), + ), + }), + ); + yield* Effect.addFinalizer(() => Deferred.succeed(releaseDetection, true)); + const base = { + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + turnId: asTurnId("blocked-diff-turn"), + createdAt: "2026-01-01T00:00:00.000Z", + }; + yield* Effect.promise(() => + harness.emitAndDrain([ + { ...base, type: "turn.started", eventId: asEventId("evt-blocked-turn-start") }, + ]), + ); + harness.emit({ + ...base, + type: "turn.diff.updated", + eventId: asEventId("evt-blocked-diff"), + payload: { unifiedDiff: "diff --git a/file.ts b/file.ts\n+new\n" }, + }); + yield* Deferred.await(detectionStarted); + + const settled = yield* harness.engine.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "thread.session-set" && + event.payload.threadId === base.threadId && + event.payload.session.status === "ready", + ), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ); + harness.emit({ + ...base, + type: "item.completed", + eventId: asEventId("evt-blocked-final-reply"), + itemId: asItemId("blocked-final-reply"), + payload: { itemType: "assistant_message", status: "completed", detail: "Work finished." }, + }); + harness.emit({ + ...base, + type: "turn.completed", + eventId: asEventId("evt-blocked-turn-completed"), + payload: { state: "completed" }, + }); + // Resolves only if turn.completed is processed while detection is still blocked. + yield* Fiber.join(settled); + const blocked = yield* Effect.promise(harness.readModel); + expect(blocked.threads[0]?.session).toMatchObject({ status: "ready", activeTurnId: null }); + expect(blocked.threads[0]?.messages).toEqual( + expect.arrayContaining([expect.objectContaining({ text: "Work finished." })]), + ); + expect(blocked.threads[0]?.checkpoints).toEqual([]); + + yield* Deferred.succeed(releaseDetection, true); + yield* Effect.promise(harness.drain); + const released = yield* Effect.promise(harness.readModel); + expect(released.threads[0]?.checkpoints).toEqual([ + expect.objectContaining({ turnId: "blocked-diff-turn", status: "missing" }), + ]); + expect(released.threads[0]?.session?.status).toBe("ready"); + }), + ); + effectIt.effect("tracks provider diff updates from a nested Git workspace", () => Effect.gen(function* () { const harness = yield* Effect.promise(() => diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index ee4d08c36d5a..154dd94a9147 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2287,12 +2287,20 @@ const make = Effect.gen(function* () { ); const worker = yield* makeDrainableWorker(processInputSafely); + // Diff bookkeeping detects the Git repository through VCS subprocesses, + // which can stall behind slow or hung git. Keep it off the worker that + // persists messages and settles turns so a stuck diff never blocks + // turn.completed. + const diffWorker = yield* makeDrainableWorker(processInputSafely); const start: ProviderRuntimeIngestionShape["start"] = () => Effect.gen(function* () { yield* forkParked( Stream.runForEach(providerService.streamEvents, (event) => - worker.enqueue({ source: "runtime", event }), + (event.type === "turn.diff.updated" ? diffWorker : worker).enqueue({ + source: "runtime", + event, + }), ), ); yield* forkParked( @@ -2307,7 +2315,7 @@ const make = Effect.gen(function* () { return { start, - drain: worker.drain, + drain: worker.drain.pipe(Effect.andThen(diffWorker.drain)), } satisfies ProviderRuntimeIngestionShape; }); From 998c733a68585052c730615389e510184a695c74 Mon Sep 17 00:00:00 2001 From: Patrik Votocek Date: Tue, 15 Sep 2026 23:19:26 +0200 Subject: [PATCH 2/4] fix(server): drop late diff placeholders once the turn has settled With diff bookkeeping on its own worker, repository detection can return after the turn completed or a newer turn started. The placeholder checkpoint then rewrote a failed turn as completed and moved the latest-turn pointer back to the old turn. Re-read the turn after detection and skip the placeholder unless it is still running. The checkpoint reactor owns the checkpoint after that. Co-Authored-By: Claude Fable 5.1 --- .../Layers/ProviderRuntimeIngestion.test.ts | 47 ++++++++++++++++--- .../Layers/ProviderRuntimeIngestion.ts | 8 ++++ 2 files changed, 48 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 3adcd79e17ed..c68fb53a752f 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -43,6 +43,8 @@ import { afterEach, describe, expect, it } from "vite-plus/test"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; +import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts"; +import { ProjectionTurnRepository } from "../../persistence/Services/ProjectionTurns.ts"; import { ProviderService, type ProviderServiceShape, @@ -415,6 +417,12 @@ describe("ProviderRuntimeIngestion", () => { engine, dispatch, readModel: () => testRuntime.runPromise(snapshotQuery.getSnapshot()), + readTurn: (turnId: TurnId) => + testRuntime.runPromise( + Effect.flatMap(ProjectionTurnRepository, (turns) => + turns.getByTurnId({ threadId: asThreadId("thread-1"), turnId }), + ).pipe(Effect.map(Option.getOrUndefined), Effect.provide(ProjectionTurnRepositoryLive)), + ), readThreadShell: () => testRuntime.runPromise( snapshotQuery @@ -3796,7 +3804,7 @@ describe("ProviderRuntimeIngestion", () => { (event) => event.type === "thread.session-set" && event.payload.threadId === base.threadId && - event.payload.session.status === "ready", + event.payload.session.status === "error", ), Stream.runHead, Effect.forkScoped({ startImmediately: true }), @@ -3812,24 +3820,49 @@ describe("ProviderRuntimeIngestion", () => { ...base, type: "turn.completed", eventId: asEventId("evt-blocked-turn-completed"), - payload: { state: "completed" }, + payload: { state: "failed" }, }); // Resolves only if turn.completed is processed while detection is still blocked. yield* Fiber.join(settled); const blocked = yield* Effect.promise(harness.readModel); - expect(blocked.threads[0]?.session).toMatchObject({ status: "ready", activeTurnId: null }); + expect(blocked.threads[0]?.session).toMatchObject({ status: "error", activeTurnId: null }); expect(blocked.threads[0]?.messages).toEqual( expect.arrayContaining([expect.objectContaining({ text: "Work finished." })]), ); expect(blocked.threads[0]?.checkpoints).toEqual([]); + // A newer turn starts before detection returns. The late placeholder + // must neither settle the failed turn as completed nor move the + // latest-turn pointer back to it. + const nextTurnId = asTurnId("next-turn"); + const nextTurnStarted = yield* harness.engine.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "thread.session-set" && + event.payload.session.activeTurnId === nextTurnId, + ), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ); + harness.emit({ + ...base, + type: "turn.started", + turnId: nextTurnId, + eventId: asEventId("evt-next-turn-start"), + }); + yield* Fiber.join(nextTurnStarted); yield* Deferred.succeed(releaseDetection, true); yield* Effect.promise(harness.drain); const released = yield* Effect.promise(harness.readModel); - expect(released.threads[0]?.checkpoints).toEqual([ - expect.objectContaining({ turnId: "blocked-diff-turn", status: "missing" }), - ]); - expect(released.threads[0]?.session?.status).toBe("ready"); + expect(released.threads[0]?.checkpoints).toEqual([]); + expect(released.threads[0]?.latestTurn).toMatchObject({ + turnId: nextTurnId, + state: "running", + }); + expect(yield* Effect.promise(() => harness.readTurn(base.turnId))).toMatchObject({ + state: "error", + checkpointRef: null, + }); }), ); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 154dd94a9147..6fdd2fca0676 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2093,6 +2093,14 @@ const make = Effect.gen(function* () { workspaceCwd && (yield* checkpointStore.isGitRepository(workspaceCwd)) ) { + // Repository detection runs off the lifecycle worker and can return + // long after the turn ended. The placeholder only marks work in + // progress; a late one would rewrite the settled turn's state and + // move the latest-turn pointer back, so drop it once the turn is over. + const turn = yield* projectionTurnRepository.getByTurnId({ threadId: thread.id, turnId }); + if (Option.isSome(turn) && turn.value.state !== "running") { + return; + } // Skip if a checkpoint already exists for this turn. A real // (non-placeholder) capture from CheckpointReactor should not // be clobbered, and dispatching a duplicate placeholder for the From b28afe8a23c3cccc9955e4b770d2d69d8fcbbf56 Mon Sep 17 00:00:00 2001 From: Patrik Votocek Date: Tue, 15 Sep 2026 23:25:17 +0200 Subject: [PATCH 3/4] fix(server): order the diff placeholder guard with turn lifecycle events Reading the turn state on the diff worker and then dispatching left a window where the lifecycle worker could settle the turn in between, so the placeholder still landed on a finished turn. The diff worker now only performs repository detection and hands the confirmed diff back to the lifecycle worker, which runs the running-turn check and the dispatch in order with the turn's terminal events. Co-Authored-By: Claude Fable 5.1 --- .../Layers/ProviderRuntimeIngestion.ts | 173 ++++++++++-------- 1 file changed, 96 insertions(+), 77 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 6fdd2fca0676..0951aebfdfe2 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -121,6 +121,8 @@ type TurnStartRequestedDomainEvent = Extract< { type: "thread.turn-start-requested" } >; +type ProviderDiffEvent = Extract; + type RuntimeIngestionInput = | { source: "runtime"; @@ -129,6 +131,11 @@ type RuntimeIngestionInput = | { source: "domain"; event: TurnStartRequestedDomainEvent; + } + | { + /** A diff whose workspace the diff worker confirmed is a Git repository. */ + source: "diff"; + event: ProviderDiffEvent; }; function toTurnId(value: TurnId | string | undefined): TurnId | undefined { @@ -2078,56 +2085,6 @@ const make = Effect.gen(function* () { } } - if (event.type === "turn.diff.updated") { - const turnId = toTurnId(event.turnId); - const checkpointContext = turnId - ? yield* projectionSnapshotQuery - .getThreadCheckpointContext(thread.id) - .pipe(Effect.map(Option.getOrUndefined)) - : undefined; - const workspaceCwd = - checkpointContext?.worktreePath ?? checkpointContext?.workspaceRoot ?? undefined; - if ( - turnId && - checkpointContext && - workspaceCwd && - (yield* checkpointStore.isGitRepository(workspaceCwd)) - ) { - // Repository detection runs off the lifecycle worker and can return - // long after the turn ended. The placeholder only marks work in - // progress; a late one would rewrite the settled turn's state and - // move the latest-turn pointer back, so drop it once the turn is over. - const turn = yield* projectionTurnRepository.getByTurnId({ threadId: thread.id, turnId }); - if (Option.isSome(turn) && turn.value.state !== "running") { - return; - } - // Skip if a checkpoint already exists for this turn. A real - // (non-placeholder) capture from CheckpointReactor should not - // be clobbered, and dispatching a duplicate placeholder for the - // same turnId would produce an unstable checkpointTurnCount. - if (hasCheckpointForTurn(checkpointContext.checkpoints, turnId)) { - // Already tracked; no-op. - } else { - const assistantMessageId = MessageId.make( - `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, - ); - yield* orchestrationEngine.dispatch({ - type: "thread.turn.diff.complete", - commandId: yield* providerCommandId(event, "thread-turn-diff-complete"), - threadId: thread.id, - turnId, - completedAt: now, - checkpointRef: CheckpointRef.make(`provider-diff:${event.eventId}`), - status: "missing", - files: [], - assistantMessageId, - checkpointTurnCount: maxCheckpointTurnCount(checkpointContext.checkpoints) + 1, - createdAt: now, - }); - } - } - } - if (event.type === "task.started" || event.type === "task.progress") { const description = event.payload.description?.trim(); if (description) { @@ -2276,39 +2233,100 @@ const make = Effect.gen(function* () { const processDomainEvent = (_event: TurnStartRequestedDomainEvent) => Effect.void; - const processInput = (input: RuntimeIngestionInput) => - input.source === "runtime" ? processRuntimeEvent(input.event) : processDomainEvent(input.event); + // Records a mid-turn placeholder checkpoint for a provider diff. Runs on the + // lifecycle worker, after repository detection, so the running-turn check + // and the dispatch are ordered with the turn's terminal events: a diff that + // resolved after turn.completed must not rewrite the settled turn's state or + // move the latest-turn pointer back. + const recordProviderDiff = Effect.fn("recordProviderDiff")(function* (event: ProviderDiffEvent) { + const thread = yield* resolveThreadRuntimeContext(event.threadId); + const turnId = toTurnId(event.turnId); + if (!thread || !turnId) return; + const turn = yield* projectionTurnRepository.getByTurnId({ threadId: thread.id, turnId }); + if (Option.isSome(turn) && turn.value.state !== "running") return; + const checkpointContext = yield* projectionSnapshotQuery + .getThreadCheckpointContext(thread.id) + .pipe(Effect.map(Option.getOrUndefined)); + // Skip if a checkpoint already exists for this turn. A real + // (non-placeholder) capture from CheckpointReactor should not + // be clobbered, and dispatching a duplicate placeholder for the + // same turnId would produce an unstable checkpointTurnCount. + if (!checkpointContext || hasCheckpointForTurn(checkpointContext.checkpoints, turnId)) return; + const now = event.createdAt; + yield* orchestrationEngine.dispatch({ + type: "thread.turn.diff.complete", + commandId: yield* providerCommandId(event, "thread-turn-diff-complete"), + threadId: thread.id, + turnId, + completedAt: now, + checkpointRef: CheckpointRef.make(`provider-diff:${event.eventId}`), + status: "missing", + files: [], + assistantMessageId: MessageId.make( + `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, + ), + checkpointTurnCount: maxCheckpointTurnCount(checkpointContext.checkpoints) + 1, + createdAt: now, + }); + }); - const processInputSafely = (input: RuntimeIngestionInput) => - processInput(input).pipe( - Effect.catchCause((cause) => { - if (Cause.hasInterruptsOnly(cause)) { - return Effect.failCause(cause); - } - return Effect.logWarning("provider runtime ingestion failed to process event", { - source: input.source, - eventId: input.event.eventId, - eventType: input.event.type, - cause: Cause.pretty(cause), - }); - }), - ); + const processInput = (input: RuntimeIngestionInput) => { + switch (input.source) { + case "runtime": + return processRuntimeEvent(input.event); + case "domain": + return processDomainEvent(input.event); + case "diff": + return recordProviderDiff(input.event); + } + }; + + const logIngestionFailure = + (source: string, event: { readonly eventId: string; readonly type: string }) => + (effect: Effect.Effect) => + effect.pipe( + Effect.catchCause((cause) => { + if (Cause.hasInterruptsOnly(cause)) { + return Effect.failCause(cause); + } + return Effect.logWarning("provider runtime ingestion failed to process event", { + source, + eventId: event.eventId, + eventType: event.type, + cause: Cause.pretty(cause), + }); + }), + ); - const worker = yield* makeDrainableWorker(processInputSafely); - // Diff bookkeeping detects the Git repository through VCS subprocesses, - // which can stall behind slow or hung git. Keep it off the worker that - // persists messages and settles turns so a stuck diff never blocks - // turn.completed. - const diffWorker = yield* makeDrainableWorker(processInputSafely); + const worker = yield* makeDrainableWorker((input: RuntimeIngestionInput) => + processInput(input).pipe(logIngestionFailure(input.source, input.event)), + ); + + // Repository detection for a diff goes through VCS subprocesses, which can + // stall behind slow or hung git. It runs on its own worker so a stuck diff + // never delays the lifecycle worker; confirmed diffs are handed back to it. + const detectProviderDiffRepository = Effect.fn("detectProviderDiffRepository")(function* ( + event: ProviderDiffEvent, + ) { + if (!toTurnId(event.turnId)) return; + const checkpointContext = yield* projectionSnapshotQuery + .getThreadCheckpointContext(event.threadId) + .pipe(Effect.map(Option.getOrUndefined)); + const workspaceCwd = checkpointContext?.worktreePath ?? checkpointContext?.workspaceRoot; + if (!workspaceCwd || !(yield* checkpointStore.isGitRepository(workspaceCwd))) return; + yield* worker.enqueue({ source: "diff", event }); + }); + const diffWorker = yield* makeDrainableWorker((event: ProviderDiffEvent) => + detectProviderDiffRepository(event).pipe(logIngestionFailure("diff", event)), + ); const start: ProviderRuntimeIngestionShape["start"] = () => Effect.gen(function* () { yield* forkParked( Stream.runForEach(providerService.streamEvents, (event) => - (event.type === "turn.diff.updated" ? diffWorker : worker).enqueue({ - source: "runtime", - event, - }), + event.type === "turn.diff.updated" + ? diffWorker.enqueue(event) + : worker.enqueue({ source: "runtime", event }), ), ); yield* forkParked( @@ -2323,7 +2341,8 @@ const make = Effect.gen(function* () { return { start, - drain: worker.drain.pipe(Effect.andThen(diffWorker.drain)), + // The diff worker feeds the lifecycle worker, so drain it first. + drain: diffWorker.drain.pipe(Effect.andThen(worker.drain)), } satisfies ProviderRuntimeIngestionShape; }); From 8a51dd877aa69a20f5497b258e545aba4cb2f4bd Mon Sep 17 00:00:00 2001 From: Patrik Votocek Date: Wed, 16 Sep 2026 00:24:25 +0200 Subject: [PATCH 4/4] fix(server): reject diff placeholders once a checkpoint was captured Provider diff ingestion checks for an existing checkpoint before dispatching a placeholder, but CheckpointReactor can commit the real capture between that read and the dispatch. Enforce the rule in the decider, which runs under the engine's command lock, so a late placeholder can no longer clobber a captured checkpoint in the SQL projection. Co-Authored-By: Claude Fable 5.1 --- apps/server/src/orchestration/decider.ts | 20 ++- .../decider.turnDiffComplete.test.ts | 127 ++++++++++++++++++ 2 files changed, 146 insertions(+), 1 deletion(-) create mode 100644 apps/server/src/orchestration/decider.turnDiffComplete.test.ts diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 86be0610f804..22561ed84fe4 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -2062,11 +2062,29 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.turn.diff.complete": { - yield* requireThread({ + const thread = yield* requireThread({ readModel, command, threadId: command.threadId, }); + // A placeholder (status "missing") must never replace a checkpoint that + // was already captured with a real git ref. Provider diff ingestion + // checks this before dispatching, but CheckpointReactor can commit the + // real capture in between; the decider runs under the engine's command + // lock, so rejecting here closes that window. + const existingCheckpoint = thread.checkpoints.find( + (checkpoint) => checkpoint.turnId === command.turnId, + ); + if ( + command.status === "missing" && + existingCheckpoint !== undefined && + existingCheckpoint.status !== "missing" + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `turn ${command.turnId} already has a captured checkpoint`, + }); + } return { ...(yield* withEventBase({ aggregateKind: "thread", diff --git a/apps/server/src/orchestration/decider.turnDiffComplete.test.ts b/apps/server/src/orchestration/decider.turnDiffComplete.test.ts new file mode 100644 index 000000000000..fa52c4eaa044 --- /dev/null +++ b/apps/server/src/orchestration/decider.turnDiffComplete.test.ts @@ -0,0 +1,127 @@ +import { + CheckpointRef, + CommandId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationCheckpointSummary, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +const NOW = "2026-01-01T00:00:00.000Z"; +const THREAD_ID = ThreadId.make("thread-1"); +const TURN_ID = TurnId.make("turn-1"); + +function makeReadModel(checkpoints: ReadonlyArray) { + return { + snapshotSequence: 0, + projects: [], + threads: [ + { + id: THREAD_ID, + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + pullRequests: [], + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + snoozedUntil: null, + snoozedAt: null, + pinnedAt: null, + pinOrderKey: null, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints, + session: null, + }, + ], + updatedAt: NOW, + } satisfies OrchestrationReadModel; +} + +function makeCheckpoint(status: OrchestrationCheckpointSummary["status"]) { + return { + turnId: TURN_ID, + checkpointTurnCount: 1, + checkpointRef: CheckpointRef.make(`existing:${status}`), + status, + files: [], + assistantMessageId: null, + completedAt: NOW, + } satisfies OrchestrationCheckpointSummary; +} + +function placeholderCommand() { + return { + type: "thread.turn.diff.complete", + commandId: CommandId.make("cmd-diff-placeholder"), + threadId: THREAD_ID, + turnId: TURN_ID, + completedAt: NOW, + checkpointRef: CheckpointRef.make("provider-diff:event-1"), + status: "missing", + files: [], + assistantMessageId: MessageId.make("assistant:turn-1"), + checkpointTurnCount: 2, + createdAt: NOW, + } as const; +} + +it.layer(NodeServices.layer)("turn diff complete decider", (it) => { + it.effect("rejects a placeholder when the turn already has a captured checkpoint", () => + Effect.gen(function* () { + const exit = yield* Effect.exit( + decideOrchestrationCommand({ + command: placeholderCommand(), + readModel: makeReadModel([makeCheckpoint("ready")]), + }), + ); + expect(Exit.isFailure(exit)).toBe(true); + }), + ); + + it.effect("accepts a placeholder when the turn only has a placeholder", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: placeholderCommand(), + readModel: makeReadModel([makeCheckpoint("missing")]), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events).toHaveLength(1); + expect(events[0]?.type).toBe("thread.turn-diff-completed"); + }), + ); + + it.effect("lets a captured checkpoint replace an earlier placeholder", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + ...placeholderCommand(), + checkpointRef: CheckpointRef.make("refs/t3/checkpoints/turn-1"), + status: "ready", + }, + readModel: makeReadModel([makeCheckpoint("missing")]), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events[0]?.type).toBe("thread.turn-diff-completed"); + }), + ); +});