diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 062a91f8cb53..c68fb53a752f 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"; @@ -42,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, @@ -267,6 +270,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 +331,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), @@ -405,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 @@ -3749,6 +3767,105 @@ 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 === "error", + ), + 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: "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: "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(released.threads[0]?.latestTurn).toMatchObject({ + turnId: nextTurnId, + state: "running", + }); + expect(yield* Effect.promise(() => harness.readTurn(base.turnId))).toMatchObject({ + state: "error", + checkpointRef: null, + }); + }), + ); + 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..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,48 +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)) - ) { - // 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) { @@ -2268,31 +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 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 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 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((input: RuntimeIngestionInput) => + processInput(input).pipe(logIngestionFailure(input.source, input.event)), + ); - const worker = yield* makeDrainableWorker(processInputSafely); + // 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) => - worker.enqueue({ source: "runtime", event }), + event.type === "turn.diff.updated" + ? diffWorker.enqueue(event) + : worker.enqueue({ source: "runtime", event }), ), ); yield* forkParked( @@ -2307,7 +2341,8 @@ const make = Effect.gen(function* () { return { start, - drain: worker.drain, + // The diff worker feeds the lifecycle worker, so drain it first. + drain: diffWorker.drain.pipe(Effect.andThen(worker.drain)), } satisfies ProviderRuntimeIngestionShape; }); 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"); + }), + ); +});