diff --git a/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts b/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts index ff19634917f31..0a47aa12e4b21 100644 --- a/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts +++ b/apps/server/src/orchestration-v2/CheckpointCaptureService.test.ts @@ -16,6 +16,11 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as Exit from "effect/Exit"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; +import * as EventStore from "./EventStore.ts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -269,6 +274,7 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { Layer.provide( Layer.mergeAll( IdAllocator.layer, + ThreadCommandExecutor.layer, refLookupFails ? CheckpointService.layer.pipe( Layer.provide( @@ -300,6 +306,10 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { capture: () => Effect.succeed(captured), }), Layer.mock(EventSink.EventSinkV2)({ + writeIfRunCurrent: (input) => + Ref.set(committed, input.events).pipe( + Effect.as({ committed: true, storedEvents: [] }), + ), commitCommand: (input) => Ref.set(committed, input.events).pipe( Effect.as({ @@ -448,6 +458,7 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { Layer.provide( Layer.mergeAll( IdAllocator.layer, + ThreadCommandExecutor.layer, Layer.mock(CheckpointService.CheckpointServiceV2)({ materializeBaselineCheckpoint: () => Effect.die("a discarded run has no baseline"), capture: () => Effect.die("a discarded run must not be captured"), @@ -455,6 +466,8 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { Layer.mock(EventSink.EventSinkV2)({ commitCommand: (input) => Ref.set(committed, input.events).pipe(Effect.as({ committed: true } as never)), + writeIfRunCurrent: (input) => + Ref.set(committed, input.events).pipe(Effect.as({ committed: true } as never)), }), ), ), @@ -477,17 +490,17 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { }), ); - it.effect("records the checkpoint of a cancelled run and keeps it cancelled", () => + const seedCaptureRun = (key: string, terminalStatus: "waiting" | "cancelled" | "interrupted") => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; const cancelledAt = DateTime.add(now, { seconds: 1 }); - const cancelledThreadId = ThreadId.make("thread:checkpoint-capture-cancelled"); - const cancelledRunId = RunId.make("run:checkpoint-capture-cancelled"); - const cancelledScopeId = CheckpointScopeId.make("scope:checkpoint-capture-cancelled"); - const cancelledRootNodeId = NodeId.make("node:checkpoint-capture-cancelled-root"); + const cancelledThreadId = ThreadId.make(`thread:checkpoint-capture:${key}`); + const cancelledRunId = RunId.make(`run:checkpoint-capture:${key}`); + const cancelledScopeId = CheckpointScopeId.make(`scope:checkpoint-capture:${key}`); + const cancelledRootNodeId = NodeId.make(`node:checkpoint-capture-root:${key}`); const cancelledProviderThreadId = ProviderThreadId.make( - "provider-thread:checkpoint-capture-cancelled", + `provider-thread:checkpoint-capture:${key}`, ); yield* projectionStore.apply({ id: EventId.make("event:checkpoint-capture-cancelled:thread"), @@ -629,7 +642,7 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { }); for (const [status, completedAt] of [ ["running", null], - ["cancelled", cancelledAt], + [terminalStatus, terminalStatus === "waiting" ? null : cancelledAt], ] as const) { yield* projectionStore.apply({ id: EventId.make(`event:checkpoint-capture-cancelled:run-${status}`), @@ -667,56 +680,290 @@ it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => { files: [], capturedAt: cancelledAt, }; - const commits = yield* Ref.make(0); - const layerCapture = CheckpointCaptureService.layer.pipe( - Layer.provide( - Layer.mergeAll( - IdAllocator.layer, - Layer.mock(CheckpointService.CheckpointServiceV2)({ - materializeBaselineCheckpoint: () => - Effect.die("baseline materialization must be skipped when ordinal 0 is ready"), - capture: () => Effect.succeed(captured), - }), - Layer.mock(EventSink.EventSinkV2)({ - commitCommand: (input) => - Effect.forEach(input.events, (event) => projectionStore.apply(event)).pipe( - Effect.andThen(Ref.update(commits, (count) => count + 1)), - Effect.as({ committed: true } as never), - Effect.orDie, - ), - }), + + return { + projectionStore, + cancelledAt, + cancelledThreadId, + cancelledRunId, + cancelledScopeId, + captured, + input: { threadId: cancelledThreadId, runId: cancelledRunId, scopeId: cancelledScopeId }, + }; + }); + + it.effect.each(["cancelled", "interrupted"] as const)( + "records the checkpoint of a %s run and preserves its terminal status on redelivery", + (status) => + Effect.gen(function* () { + const { + projectionStore, + cancelledAt, + cancelledThreadId, + cancelledRunId, + cancelledScopeId, + captured, + } = yield* seedCaptureRun(status, status); + const commits = yield* Ref.make(0); + const layerCapture = CheckpointCaptureService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + ThreadCommandExecutor.layer, + Layer.mock(CheckpointService.CheckpointServiceV2)({ + materializeBaselineCheckpoint: () => + Effect.die("baseline materialization must be skipped when ordinal 0 is ready"), + capture: () => Effect.succeed(captured), + }), + Layer.mock(EventSink.EventSinkV2)({ + writeIfRunCurrent: (input) => + Effect.forEach(input.events, (event) => projectionStore.apply(event)).pipe( + Effect.andThen(Ref.update(commits, (count) => count + 1)), + Effect.as({ committed: true, storedEvents: [] }), + Effect.orDie, + ), + commitCommand: (input) => + Effect.forEach(input.events, (event) => projectionStore.apply(event)).pipe( + Effect.andThen(Ref.update(commits, (count) => count + 1)), + Effect.as({ committed: true } as never), + Effect.orDie, + ), + }), + ), ), - ), - ); + ); - yield* Effect.gen(function* () { - const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2; - yield* service.execute({ - threadId: cancelledThreadId, - runId: cancelledRunId, - scopeId: cancelledScopeId, - }); - yield* service.execute({ - threadId: cancelledThreadId, + yield* Effect.gen(function* () { + const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2; + yield* service.execute({ + threadId: cancelledThreadId, + runId: cancelledRunId, + scopeId: cancelledScopeId, + }); + yield* service.execute({ + threadId: cancelledThreadId, + runId: cancelledRunId, + scopeId: cancelledScopeId, + }); + }).pipe(Effect.provide(layerCapture)); + + assert.equal(yield* Ref.get(commits), 1); + const projected = yield* projectionStore.getCheckpointCaptureContext(cancelledThreadId, { runId: cancelledRunId, scopeId: cancelledScopeId, }); - }).pipe(Effect.provide(layerCapture)); + assert.equal(projected.run?.status, status); + assert.equal(projected.run?.checkpointId, captured.id); + const completedAt = projected.run?.completedAt; + assert.equal( + completedAt ? DateTime.formatIso(completedAt) : completedAt, + DateTime.formatIso(cancelledAt), + ); + assert.equal(projected.rootNode?.status, status); + assert.deepEqual([...projected.readyCheckpointOrdinals].toSorted(), [0, 1]); + }), + ); - assert.equal(yield* Ref.get(commits), 1); - const projected = yield* projectionStore.getCheckpointCaptureContext(cancelledThreadId, { - runId: cancelledRunId, - scopeId: cancelledScopeId, - }); - assert.equal(projected.run?.status, "cancelled"); - assert.equal(projected.run?.checkpointId, captured.id); - const completedAt = projected.run?.completedAt; - assert.equal( - completedAt ? DateTime.formatIso(completedAt) : completedAt, - DateTime.formatIso(cancelledAt), - ); - assert.equal(projected.rootNode?.status, "cancelled"); - assert.deepEqual([...projected.readyCheckpointOrdinals].toSorted(), [0, 1]); - }), + it.effect( + "only a final Git failure settles capture; retry and interruption leave it waiting", + () => + Effect.gen(function* () { + const { input, projectionStore } = yield* seedCaptureRun("git-failure", "waiting"); + const started = yield* Deferred.make(); + let blockCapture = true; + const context = projectionStore.getCheckpointCaptureContext(input.threadId, input); + yield* Effect.gen(function* () { + const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2; + const request = { ...input, willRetry: false }; + const capturing = yield* service.execute(request).pipe(Effect.forkChild); + yield* Deferred.await(started); + yield* Fiber.interrupt(capturing); + assert.equal((yield* context).run?.status, "waiting"); + blockCapture = false; + const retryRequest = { ...input, willRetry: true }; + assert.isTrue(Exit.isFailure(yield* service.execute(retryRequest).pipe(Effect.exit))); + assert.equal((yield* context).run?.status, "waiting"); + const retrying = yield* projectionStore.getThreadProjection(input.threadId); + assert.equal( + retrying.turnItems.filter((item) => item.type === "system_notice").length, + 0, + ); + assert.isTrue(Exit.isFailure(yield* service.execute(request).pipe(Effect.exit))); + }).pipe( + Effect.provide( + CheckpointCaptureService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + ThreadCommandExecutor.layer, + EventSink.layer.pipe(Layer.provide(EventStore.layer)), + Layer.mock(CheckpointService.CheckpointServiceV2)({ + capture: () => + blockCapture + ? Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.fail( + new CheckpointService.CheckpointCaptureError({ + scopeId: input.scopeId, + cause: "Git capture failed", + }), + ), + }), + ), + ), + ), + ), + ); + const failed = yield* context; + assert.equal(failed.run?.status, "failed"); + assert.equal(failed.rootNode?.status, "failed"); + assert.isNull(failed.run?.checkpointId); + const projection = yield* projectionStore.getThreadProjection(input.threadId); + assert.lengthOf( + projection.turnItems.filter((item) => item.type === "system_notice"), + 1, + ); + }), + ); + + it.effect.each(["capture-first", "stop-first", "failure-after-stop"] as const)( + "settles checkpoint work against Stop under the thread lock: %s", + (order) => + Effect.gen(function* () { + const { input, captured, projectionStore, cancelledAt } = yield* seedCaptureRun( + order, + "waiting", + ); + const sink = yield* EventSink.EventSinkV2; + const commands = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const gitStarted = yield* Deferred.make(); + const releaseGit = yield* Deferred.make(); + const settlementAttempted = yield* Deferred.make(); + const writeEntered = yield* Deferred.make(); + const releaseWrite = yield* Deferred.make(); + const stopRead = yield* Deferred.make(); + const releaseStop = yield* Deferred.make(); + const stopAttempted = yield* Deferred.make(); + const context = projectionStore.getCheckpointCaptureContext(input.threadId, input); + const checkpointFailure = new CheckpointService.CheckpointCaptureError({ + scopeId: input.scopeId, + cause: "Git capture failed", + }); + const beforeWrite = Deferred.succeed(settlementAttempted, undefined).pipe( + Effect.andThen(Deferred.succeed(writeEntered, undefined)), + Effect.andThen(order === "capture-first" ? Deferred.await(releaseWrite) : Effect.void), + ); + const layerCapture = CheckpointCaptureService.layer.pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + Layer.succeed(ThreadCommandExecutor.ThreadCommandExecutor, { + ...commands, + withLock: (key, effect) => + Deferred.succeed(settlementAttempted, undefined).pipe( + Effect.andThen(commands.withLock(key, effect)), + ), + }), + Layer.mock(CheckpointService.CheckpointServiceV2)({ + capture: () => + Deferred.succeed(gitStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseGit)), + Effect.andThen( + order === "failure-after-stop" + ? Effect.fail(checkpointFailure) + : Effect.succeed(captured), + ), + ), + }), + Layer.succeed(EventSink.EventSinkV2, { + ...sink, + commitCommand: (request) => + beforeWrite.pipe(Effect.andThen(sink.commitCommand(request))), + writeIfRunCurrent: (request) => + beforeWrite.pipe(Effect.andThen(sink.writeIfRunCurrent(request))), + }), + ), + ), + ); + const stop = Deferred.succeed(stopAttempted, undefined).pipe( + Effect.andThen( + commands.withLock( + input.threadId, + Effect.gen(function* () { + const { run, rootNode } = yield* context; + yield* Deferred.succeed(stopRead, undefined); + if (run?.status !== "waiting" || rootNode === undefined) return; + yield* Deferred.await(releaseStop); + yield* sink.write({ + events: [ + { + id: EventId.make(`stop-run:${input.runId}`), + type: "run.updated", + threadId: input.threadId, + runId: run.id, + occurredAt: cancelledAt, + payload: { ...run, status: "interrupted", completedAt: cancelledAt }, + }, + { + id: EventId.make(`stop-node:${input.runId}`), + type: "node.updated", + threadId: input.threadId, + runId: run.id, + occurredAt: cancelledAt, + payload: { ...rootNode, status: "interrupted", completedAt: cancelledAt }, + }, + ], + }); + }), + ), + ), + ); + yield* Effect.gen(function* () { + const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2; + const request = { ...input, willRetry: false }; + const capturing = yield* service.execute(request).pipe(Effect.exit, Effect.forkChild); + yield* Deferred.await(gitStarted); + if (order === "capture-first") { + yield* Deferred.succeed(releaseGit, undefined); + yield* Deferred.await(writeEntered); + const stopping = yield* stop.pipe(Effect.forkChild); + yield* Deferred.await(stopAttempted); + yield* Deferred.succeed(releaseStop, undefined); + yield* Deferred.succeed(releaseWrite, undefined); + yield* Fiber.join(capturing); + yield* Fiber.join(stopping); + } else { + const stopping = yield* stop.pipe(Effect.forkChild); + yield* Deferred.await(stopRead); + yield* Deferred.succeed(releaseGit, undefined); + // A failed capture may return without attempting a settlement. + yield* Effect.raceFirst(Deferred.await(settlementAttempted), Fiber.await(capturing)); + assert.equal((yield* context).run?.status, "waiting"); + yield* Deferred.succeed(releaseStop, undefined); + yield* Fiber.join(stopping); + const result = yield* Fiber.join(capturing); + assert.equal(Exit.isFailure(result), order === "failure-after-stop"); + } + }).pipe(Effect.provide(layerCapture)); + const result = yield* context; + assert.equal(result.run?.status, order === "capture-first" ? "completed" : "interrupted"); + assert.equal(result.rootNode?.status, result.run?.status); + assert.equal(result.run?.checkpointId, order === "capture-first" ? captured.id : null); + const projection = yield* projectionStore.getThreadProjection(input.threadId); + assert.equal( + projection.turnItems.filter((item) => item.type === "checkpoint").length, + order === "capture-first" ? 1 : 0, + ); + assert.equal( + projection.turnItems.filter((item) => item.type === "system_notice").length, + 0, + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + ThreadCommandExecutor.layer, + EventSink.layer.pipe(Layer.provide(EventStore.layer)), + ), + ), + ), ); }); diff --git a/apps/server/src/orchestration-v2/CheckpointCaptureService.ts b/apps/server/src/orchestration-v2/CheckpointCaptureService.ts index 45bf03610316d..c4ae71f6ac55c 100644 --- a/apps/server/src/orchestration-v2/CheckpointCaptureService.ts +++ b/apps/server/src/orchestration-v2/CheckpointCaptureService.ts @@ -2,6 +2,7 @@ import { CheckpointScopeId, CommandId, type OrchestrationV2Checkpoint, + type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, type OrchestrationV2Run, @@ -9,6 +10,7 @@ import { RunId, ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -19,6 +21,7 @@ import * as CheckpointService from "./CheckpointService.ts"; import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; export class CheckpointCaptureExecutionError extends Schema.TaggedError()( "CheckpointCaptureExecutionError", @@ -37,6 +40,7 @@ export interface CheckpointCaptureServiceV2Shape { readonly threadId: ThreadId; readonly runId: RunId; readonly scopeId: CheckpointScopeId; + readonly willRetry?: boolean; }) => Effect.Effect; } @@ -52,6 +56,7 @@ export const layer: Layer.Layer< | EventSink.EventSinkV2 | IdAllocator.IdAllocatorV2 | ProjectionStore.ProjectionStoreV2 + | ThreadCommandExecutor.ThreadCommandExecutor > = Layer.effect( CheckpointCaptureServiceV2, Effect.gen(function* () { @@ -59,6 +64,7 @@ export const layer: Layer.Layer< const eventSink = yield* EventSink.EventSinkV2; const ids = yield* IdAllocator.IdAllocatorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const execute = Effect.fn("orchestrationV2.checkpointCapture.execute")(function* (input: { readonly threadId: ThreadId; @@ -134,113 +140,197 @@ export const layer: Layer.Layer< // (ProjectionStore preserves the field when absent from the payload). const { delegatedCompletion: _delegatedCompletion, ...runWithoutDelegatedCompletion } = run; const commandId = CommandId.make(`command:effect:checkpoint.capture:${run.id}`); - yield* eventSink.commitCommand({ - commandId, - threadId: input.threadId, - commandType: "checkpoint.capture", - acceptedAt: capturedAt, - effects: [], - events: [ - ...(threadStartCheckpoint === null - ? [] - : [ - { - id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), - type: "checkpoint.captured" as const, - threadId: input.threadId, - nodeId: threadStartCheckpoint.nodeId, - driver: providerThread.driver, - providerInstanceId: run.providerInstanceId, - occurredAt: capturedAt, - payload: threadStartCheckpoint, - }, - ]), - ...(baselineCheckpoint === null - ? [] - : [ - { - id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), - type: "checkpoint.captured" as const, - threadId: input.threadId, - nodeId: baselineCheckpoint.nodeId, - driver: providerThread.driver, - providerInstanceId: run.providerInstanceId, - occurredAt: capturedAt, - payload: baselineCheckpoint, - }, - ]), - { - id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), - type: "checkpoint.captured", - threadId: input.threadId, - runId: run.id, - nodeId: rootNode.id, - driver: providerThread.driver, - providerInstanceId: run.providerInstanceId, - occurredAt: capturedAt, - payload: checkpoint, - }, - { - id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), - type: "turn-item.updated", - threadId: input.threadId, - runId: run.id, - nodeId: rootNode.id, - driver: providerThread.driver, - providerInstanceId: run.providerInstanceId, - occurredAt: capturedAt, - payload: makeCheckpointTurnItem({ - idAllocator: ids, - run, - rootNode, - providerThread, - checkpoint, - completedAt: capturedAt, - }), - }, - { - id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), - type: "run.updated", + // Git and workspace admission stay outside the command lock so Stop can cancel them. + yield* eventSink + .writeIfRunCurrent({ + commandId, + threadId: input.threadId, + runId: run.id, + activeAttemptId: run.activeAttemptId, + expectedStatus: run.status, + effects: [], + events: [ + ...(threadStartCheckpoint === null + ? [] + : [ + { + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "checkpoint.captured" as const, + threadId: input.threadId, + nodeId: threadStartCheckpoint.nodeId, + driver: providerThread.driver, + providerInstanceId: run.providerInstanceId, + occurredAt: capturedAt, + payload: threadStartCheckpoint, + }, + ]), + ...(baselineCheckpoint === null + ? [] + : [ + { + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "checkpoint.captured" as const, + threadId: input.threadId, + nodeId: baselineCheckpoint.nodeId, + driver: providerThread.driver, + providerInstanceId: run.providerInstanceId, + occurredAt: capturedAt, + payload: baselineCheckpoint, + }, + ]), + { + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "checkpoint.captured", + threadId: input.threadId, + runId: run.id, + nodeId: rootNode.id, + driver: providerThread.driver, + providerInstanceId: run.providerInstanceId, + occurredAt: capturedAt, + payload: checkpoint, + }, + { + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "turn-item.updated", + threadId: input.threadId, + runId: run.id, + nodeId: rootNode.id, + driver: providerThread.driver, + providerInstanceId: run.providerInstanceId, + occurredAt: capturedAt, + payload: makeCheckpointTurnItem({ + idAllocator: ids, + run, + rootNode, + providerThread, + checkpoint, + completedAt: capturedAt, + }), + }, + { + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "run.updated", + threadId: input.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: capturedAt, + payload: stopped + ? { ...runWithoutDelegatedCompletion, checkpointId: checkpoint.id } + : { + ...runWithoutDelegatedCompletion, + status: "completed", + completedAt: capturedAt, + checkpointId: checkpoint.id, + }, + }, + ...(stopped + ? [] + : [ + { + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "node.updated" as const, + threadId: input.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: capturedAt, + payload: { + ...rootNode, + status: "completed" as const, + completedAt: capturedAt, + checkpointScopeId: scope.id, + }, + }, + ]), + ], + }) + .pipe((settle) => threadCommands.withLock(input.threadId, settle)); + }); + + const failCapture = Effect.fn("orchestrationV2.checkpointCapture.fail")(function* ( + input: Parameters[0], + ) { + yield* threadCommands.withLock( + input.threadId, + Effect.gen(function* () { + const { run, rootNode } = yield* projections.getCheckpointCaptureContext( + input.threadId, + input, + ); + if (run?.status !== "waiting") return; + const now = yield* DateTime.now; + const commandId = CommandId.make(`command:effect:checkpoint.capture.failed:${run.id}`); + const { delegatedCompletion: _delegatedCompletion, ...currentRun } = run; + const eventContext = { threadId: input.threadId, runId: run.id, - nodeId: rootNode.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), providerInstanceId: run.providerInstanceId, - occurredAt: capturedAt, - payload: stopped - ? { ...runWithoutDelegatedCompletion, checkpointId: checkpoint.id } - : { - ...runWithoutDelegatedCompletion, - status: "completed", - completedAt: capturedAt, - checkpointId: checkpoint.id, - }, - }, - ...(stopped - ? [] - : [ - { - id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), - type: "node.updated" as const, + occurredAt: now, + }; + const events: Array = [ + { + ...eventContext, + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "run.updated", + payload: { ...currentRun, status: "failed", completedAt: now }, + }, + ]; + if (rootNode !== undefined) { + events.push( + { + ...eventContext, + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "node.updated", + payload: { ...rootNode, status: "failed", completedAt: now }, + }, + { + ...eventContext, + id: yield* ids.allocate.event({ threadId: input.threadId, commandId }), + type: "turn-item.updated", + payload: { + id: ids.derive.runSignalTurnItem({ runId: run.id, signal: "checkpoint-failed" }), threadId: input.threadId, runId: run.id, nodeId: rootNode.id, - providerInstanceId: run.providerInstanceId, - occurredAt: capturedAt, - payload: { - ...rootNode, - status: "completed" as const, - completedAt: capturedAt, - checkpointScopeId: scope.id, - }, + providerThreadId: run.providerThreadId, + providerTurnId: rootNode.providerTurnId, + nativeItemRef: null, + parentItemId: null, + ordinal: run.ordinal * 100 + 99, + type: "system_notice", + status: "completed", + title: "Checkpoint capture failed", + message: + "Checkpoint capture failed after retries. This turn has no checkpoint to restore.", + startedAt: now, + completedAt: now, + updatedAt: now, }, - ]), - ], - }); + }, + ); + } + yield* eventSink.writeIfRunCurrent({ + commandId, + threadId: input.threadId, + runId: run.id, + activeAttemptId: run.activeAttemptId, + expectedStatus: "waiting", + events, + }); + }), + ); }); return CheckpointCaptureServiceV2.of({ execute: (input) => execute(input).pipe( + Effect.tapCause((cause) => + input.willRetry === false && !Cause.hasInterruptsOnly(cause) + ? failCapture(input) + : Effect.void, + ), Effect.mapError((cause) => isCheckpointCaptureExecutionError(cause) ? cause diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 6227e3a5df217..f091607ecf792 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -448,6 +448,7 @@ export const layerExecutor: Layer.Layer< threadId: effect.threadId, runId: effect.request.runId, scopeId: effect.request.scopeId, + willRetry, }) .pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 55c2855144a0f..cfd4d3dd3dd20 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -82,7 +82,7 @@ export interface EventSinkV2Shape { readonly commandId?: CommandId; readonly threadId: ThreadId; readonly runId: RunId; - readonly activeAttemptId: RunAttemptId; + readonly activeAttemptId: RunAttemptId | null; readonly expectedStatus: OrchestrationV2Run["status"]; readonly events: ReadonlyArray; readonly effects?: ReadonlyArray; diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 3a50b441a84ca..c6d98f8b6cd3f 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1542,136 +1542,145 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }).pipe(Effect.provide(Layer.fresh(layerTest))), ); - it.effect("does not publish a stale provider start after an interrupt wins", () => - Effect.gen(function* () { - const eventSink = yield* EventSink.EventSinkV2; - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const outbox = yield* EffectOutbox.EffectOutboxV2; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:foundation-stale-provider-start"); - const runId = RunId.make("run:foundation-stale-provider-start"); - const attemptId = RunAttemptId.make("run-attempt:foundation-stale-provider-start"); - const thread = makeThread(threadId, now); - const startingRun: OrchestrationV2Run = { - id: runId, - threadId, - ordinal: 1, - providerInstanceId, - modelSelection, - providerThreadId: null, - userMessageId: MessageId.make("message:foundation-stale-provider-start"), - rootNodeId: null, - activeAttemptId: attemptId, - status: "starting", - queuePosition: null, - requestedAt: now, - startedAt: null, - completedAt: null, - checkpointId: null, - contextHandoffId: null, - }; - yield* eventSink.write({ - events: [ - threadCreatedEvent({ - id: "event:foundation-stale-provider-start:thread", - thread, - now, - }), - { - id: EventId.make("event:foundation-stale-provider-start:run"), - type: "run.created", - threadId, - runId, - providerInstanceId, - occurredAt: now, - payload: startingRun, - }, - ], - }); - - const captureEffect = { - id: "effect:foundation-current-capture", - commandId: CommandId.make("command:foundation-current-capture"), - threadId, - request: { - type: "checkpoint.capture" as const, - runId, - scopeId: CheckpointScopeId.make("scope:foundation-current-capture"), - }, - }; - assert.isTrue( - (yield* eventSink.writeIfRunCurrent({ - threadId, - runId, - activeAttemptId: attemptId, - expectedStatus: "starting", - events: [], - effects: [captureEffect], - })).committed, - ); - assert.isTrue(Option.isSome(yield* outbox.get(captureEffect.id))); - yield* outbox.awaitAvailable; - const staleCaptureEffect = { ...captureEffect, id: "effect:foundation-stale-capture" }; - - const reachedPrecommitGap = yield* Deferred.make(); - const releaseStaleStart = yield* Deferred.make(); - const providerStartCount = yield* Ref.make(0); - const staleStartFiber = yield* Effect.gen(function* () { - yield* Deferred.succeed(reachedPrecommitGap, undefined); - yield* Deferred.await(releaseStaleStart); - const result = yield* eventSink.writeIfRunCurrent({ + it.effect.each(["starting", "waiting"] as const)( + "does not publish stale work after an interrupt wins (%s)", + (status) => + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:foundation-stale-provider-start"); + const runId = RunId.make("run:foundation-stale-provider-start"); + const attemptId = + status === "waiting" + ? null + : RunAttemptId.make("run-attempt:foundation-stale-provider-start"); + const thread = makeThread(threadId, now); + const startingRun: OrchestrationV2Run = { + id: runId, threadId, - runId, + ordinal: 1, + providerInstanceId, + modelSelection, + providerThreadId: null, + userMessageId: MessageId.make("message:foundation-stale-provider-start"), + rootNodeId: null, activeAttemptId: attemptId, - expectedStatus: "starting", - effects: [staleCaptureEffect], + status, + queuePosition: null, + requestedAt: now, + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + yield* eventSink.write({ events: [ + threadCreatedEvent({ + id: "event:foundation-stale-provider-start:thread", + thread, + now, + }), { - id: EventId.make("event:foundation-stale-provider-start:running"), - type: "run.updated", + id: EventId.make("event:foundation-stale-provider-start:run"), + type: "run.created", threadId, runId, providerInstanceId, occurredAt: now, - payload: { ...startingRun, status: "running", startedAt: now }, + payload: startingRun, }, ], }); - if (result.committed) { - yield* Ref.update(providerStartCount, (count) => count + 1); - } - return result; - }).pipe(Effect.forkChild); - yield* Deferred.await(reachedPrecommitGap); - const interruptedAt = yield* DateTime.now; - yield* eventSink.write({ - events: [ - { - id: EventId.make("event:foundation-stale-provider-start:cancelled"), - type: "run.updated", - threadId, + const captureEffect = { + id: "effect:foundation-current-capture", + commandId: CommandId.make("command:foundation-current-capture"), + threadId, + request: { + type: "checkpoint.capture" as const, runId, - providerInstanceId, - occurredAt: interruptedAt, - payload: { - ...startingRun, - status: "cancelled", - completedAt: interruptedAt, - }, + scopeId: CheckpointScopeId.make("scope:foundation-current-capture"), }, - ], - }); - yield* Deferred.succeed(releaseStaleStart, undefined); + }; + assert.isTrue( + (yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: attemptId, + expectedStatus: status, + events: [], + effects: [captureEffect], + })).committed, + ); + assert.isTrue(Option.isSome(yield* outbox.get(captureEffect.id))); + yield* outbox.awaitAvailable; + const staleCaptureEffect = { ...captureEffect, id: "effect:foundation-stale-capture" }; + + const reachedPrecommitGap = yield* Deferred.make(); + const releaseStaleStart = yield* Deferred.make(); + const providerStartCount = yield* Ref.make(0); + const staleStartFiber = yield* Effect.gen(function* () { + yield* Deferred.succeed(reachedPrecommitGap, undefined); + yield* Deferred.await(releaseStaleStart); + const result = yield* eventSink.writeIfRunCurrent({ + threadId, + runId, + activeAttemptId: attemptId, + expectedStatus: status, + effects: [staleCaptureEffect], + events: [ + { + id: EventId.make("event:foundation-stale-provider-start:running"), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: now, + payload: { + ...startingRun, + status: status === "waiting" ? "completed" : "running", + startedAt: now, + }, + }, + ], + }); + if (result.committed) { + yield* Ref.update(providerStartCount, (count) => count + 1); + } + return result; + }).pipe(Effect.forkChild); - const staleResult = yield* Fiber.join(staleStartFiber); - assert.isFalse(staleResult.committed); - assert.isTrue(Option.isNone(yield* outbox.get(staleCaptureEffect.id))); - assert.deepEqual(staleResult.storedEvents, []); - assert.equal(yield* Ref.get(providerStartCount), 0); - const projection = yield* projectionStore.getThreadProjection(threadId); - assert.equal(projection.runs[0]?.status, "cancelled"); - }).pipe(Effect.provide(Layer.fresh(layerTest))), + yield* Deferred.await(reachedPrecommitGap); + const interruptedAt = yield* DateTime.now; + yield* eventSink.write({ + events: [ + { + id: EventId.make("event:foundation-stale-provider-start:cancelled"), + type: "run.updated", + threadId, + runId, + providerInstanceId, + occurredAt: interruptedAt, + payload: { + ...startingRun, + status: "cancelled", + completedAt: interruptedAt, + }, + }, + ], + }); + yield* Deferred.succeed(releaseStaleStart, undefined); + + const staleResult = yield* Fiber.join(staleStartFiber); + assert.isFalse(staleResult.committed); + assert.isTrue(Option.isNone(yield* outbox.get(staleCaptureEffect.id))); + assert.deepEqual(staleResult.storedEvents, []); + assert.equal(yield* Ref.get(providerStartCount), 0); + const projection = yield* projectionStore.getThreadProjection(threadId); + assert.equal(projection.runs[0]?.status, "cancelled"); + }).pipe(Effect.provide(Layer.fresh(layerTest))), ); it.effect("guards post-terminal provider-thread writes by attempt and run ordinal", () => diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 4464d09916e33..ecff803ffa7ac 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9385,6 +9385,64 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } } + // The provider has finished; Stop only cancels the local checkpoint work. + if (run.status === "waiting" && !hasBackgroundWork) { + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: interruptRequestItem, + }); + yield* emitEvent({ + type: "turn-item.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { + ...interruptRequestItem, + id: idAllocator.derive.runSignalTurnItem({ runId: run.id, signal: "interrupt-result" }), + parentItemId: interruptRequestItem.id, + ordinal: interruptRequestItem.ordinal + 1, + type: "run_interrupt_result", + status: "interrupted", + title: "Interrupted", + message: "Checkpoint capture interrupted; this turn has no checkpoint to restore.", + }, + }); + yield* emitEvent({ + type: "node.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...rootNode, status: "interrupted", completedAt: now }, + }); + yield* emitEvent({ + type: "run.updated", + threadId: command.threadId, + runId: run.id, + nodeId: rootNode.id, + providerInstanceId: run.providerInstanceId, + occurredAt: now, + payload: { ...run, status: "interrupted", completedAt: now }, + }); + yield* stopRemainingWork(); + yield* Ref.update(effects, (existing) => [...existing, ...otherProviderInterrupts]); + return { + effectTypes: ["checkpoint.capture"], + reason: `Run ${run.id} was interrupted during checkpoint capture.`, + } satisfies { + readonly effectTypes: ReadonlyArray; + readonly reason: string; + }; + } + if ( providerTurn === undefined && (run.status === "preparing" || run.status === "starting" || run.status === "running") @@ -9722,7 +9780,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (run) => run.status === "preparing" || run.status === "starting" || run.status === "running", ) ?? - (derivePendingBackgroundWork({ + (latestRun?.status === "waiting" || + derivePendingBackgroundWork({ latestRun, providerThreads: projection.providerThreads, turnItems: projection.turnItems, diff --git a/apps/server/src/orchestration-v2/RunFinalizationService.ts b/apps/server/src/orchestration-v2/RunFinalizationService.ts index b1136f6f8b04f..b275b6a9442af 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationService.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationService.ts @@ -44,6 +44,7 @@ export class RunFinalizationService extends Context.Service< readonly threadId: ThreadId; readonly runId: RunId; readonly scopeId: CheckpointScopeId; + readonly willRetry?: boolean; }) => Effect.Effect; } >()("t3/orchestration-v2/RunFinalizationService") {} diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 12db927d9c6d0..4a96e2086153d 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1,5 +1,6 @@ import { assert, it } from "@effect/vitest"; import { + CheckpointScopeId, CommandId, ContextTransferId, EventId, @@ -16,6 +17,9 @@ import { ThreadId, TurnItemId, } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Fiber from "effect/Fiber"; +import * as EventSink from "./EventSink.ts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -1334,3 +1338,198 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) ); }).pipe(Effect.provide(layerTest)), ); + +const seedCheckpointWait = (key: string) => + Effect.gen(function* () { + const threadId = ThreadId.make(`thread:checkpoint-wait:${key}`); + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const sql = yield* SqlClient.SqlClient; + yield* createWatchingThread(threadId, 1); + yield* send(threadId, "active", "start_immediately"); + yield* send(threadId, "follow-up", "queue_after_active"); + const projection = yield* orchestrator.getThreadProjection(threadId); + const run = projection.runs[0]!; + const root = projection.nodes.find((node) => node.id === run.rootNodeId)!; + const queued = projection.runs[1]!; + const now = yield* DateTime.now; + const scopeId = CheckpointScopeId.make(`scope:checkpoint-wait:${key}`); + const turnId = ProviderTurnId.make(`turn:checkpoint-wait:${key}`); + // Start from the persisted post-provider state; this suite never starts provider CLIs. + yield* sql`DELETE FROM orchestration_v2_effect_outbox WHERE thread_id = ${threadId}`; + const effectId = `effect:checkpoint-wait:${key}`; + yield* sink.writeWithEffects({ + events: [ + { + id: EventId.make("checkpoint-wait:run"), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "waiting", startedAt: now, activeAttemptId: null }, + }, + { + id: EventId.make("checkpoint-wait:root"), + type: "node.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { + ...root, + status: "waiting", + checkpointScopeId: scopeId, + providerTurnId: turnId, + }, + }, + { + id: EventId.make("checkpoint-wait:queue"), + type: "run.updated", + threadId, + runId: queued.id, + occurredAt: now, + payload: { ...queued, queueHeld: true }, + }, + { + id: EventId.make("checkpoint-wait:turn"), + type: "provider-turn.updated", + threadId, + occurredAt: now, + payload: { + id: turnId, + providerThreadId: run.providerThreadId!, + nodeId: root.id, + runAttemptId: run.activeAttemptId!, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }, + ], + effects: [ + { + id: effectId, + commandId: CommandId.make("checkpoint-wait:capture"), + threadId, + request: { type: "checkpoint.capture", runId: run.id, scopeId }, + }, + ], + }); + return { threadId, run, root, queued, effectId, turnId }; + }); + +it.effect( + "exhausted checkpoint retries settle the durable run and preserve a resumable held queue", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sql = yield* SqlClient.SqlClient; + const { threadId, run, root, queued, effectId } = yield* seedCheckpointWait("failure"); + // The missing scope exercises the same final-failure path as a Git error. + yield* worker.drain(); + const retrying = yield* orchestrator.getThreadProjection(threadId); + assert.equal(retrying.runs.find((item) => item.id === run.id)?.status, "waiting"); + assert.equal(retrying.turnItems.filter((item) => item.type === "system_notice").length, 0); + for (const delay of [100, 200, 400, 800]) { + yield* TestClock.adjust(delay); + yield* worker.drain(); + } + const failed = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + yield* sql`SELECT status, attempt_count FROM orchestration_v2_effect_outbox WHERE effect_id = ${effectId}`, + [{ status: "failed", attempt_count: 5 }], + ); + assert.equal(failed.runs.find((item) => item.id === run.id)?.status, "failed"); + assert.isNotNull(failed.runs.find((item) => item.id === run.id)?.completedAt); + assert.isNull(failed.runs.find((item) => item.id === run.id)?.checkpointId); + assert.equal(failed.nodes.find((item) => item.id === root.id)?.status, "failed"); + const notices = failed.turnItems.filter( + (item) => item.type === "system_notice" && item.runId === run.id, + ); + assert.lengthOf(notices, 1); + assert.match(notices[0]!.title ?? "", /checkpoint.*failed/i); + assert.equal(failed.runs.find((item) => item.id === queued.id)?.queueHeld, true); + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("resume:failed-capture"), + threadId, + }); + const resumed = yield* orchestrator.getThreadProjection(threadId); + assert.equal(resumed.runs.find((item) => item.id === queued.id)?.status, "starting"); + assert.notEqual(resumed.runs.find((item) => item.id === queued.id)?.queueHeld, true); + }).pipe(Effect.provide(layerTest)), +); + +it.effect.each(["run.interrupt", "thread.stop"] as const)( + "%s cancels checkpoint work without interrupting the completed provider and the held queue resumes", + (stopCommand) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sql = yield* SqlClient.SqlClient; + const { threadId, run, root, queued, effectId, turnId } = yield* seedCheckpointWait("stop"); + const started = yield* Deferred.make(); + const interrupted = yield* Deferred.make(); + yield* Effect.gen(function* () { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const capturing = yield* worker.runOnce.pipe(Effect.forkChild); + yield* Effect.raceFirst( + Deferred.await(started), + Fiber.join(capturing).pipe( + Effect.flatMap((result) => + Effect.die(`Checkpoint worker exited before capture started: ${result}`), + ), + ), + ); + yield* orchestrator.dispatch({ + type: stopCommand, + commandId: CommandId.make("stop:checkpoint"), + threadId, + runId: run.id, + holdQueue: true, + }); + const stopped = yield* orchestrator.getThreadProjection(threadId); + assert.equal(stopped.runs.find((item) => item.id === run.id)?.status, "interrupted"); + assert.equal(stopped.nodes.find((item) => item.id === root.id)?.status, "interrupted"); + assert.isNull(stopped.runs.find((item) => item.id === run.id)?.checkpointId); + assert.equal(stopped.providerTurns.find((item) => item.id === turnId)?.status, "completed"); + assert.equal(stopped.runs.find((item) => item.id === queued.id)?.queueHeld, true); + yield* Deferred.await(interrupted); + yield* Fiber.join(capturing); + assert.deepEqual( + yield* sql`SELECT status FROM orchestration_v2_effect_outbox WHERE effect_id = ${effectId}`, + [{ status: "cancelled" }], + ); + assert.deepEqual( + yield* sql`SELECT effect_id FROM orchestration_v2_effect_outbox WHERE thread_id = ${threadId} AND effect_type = 'provider-turn.interrupt'`, + [], + ); + assert.equal(stopped.turnItems.filter((item) => item.type === "system_notice").length, 0); + }).pipe( + Effect.provide( + EffectWorker.layerWithOptions({ workerId: "checkpoint-stop-test" }).pipe( + Layer.provide( + Layer.succeed(EffectWorker.OrchestrationEffectExecutorV2, { + execute: (effect) => { + assert.equal(effect.request.type, "checkpoint.capture"); + return Deferred.succeed(started, undefined).pipe( + Effect.andThen(Effect.never), + Effect.onInterrupt(() => Deferred.succeed(interrupted, undefined)), + ); + }, + }), + ), + ), + ), + ); + yield* orchestrator.dispatch({ + type: "queue.resume", + commandId: CommandId.make("resume:stopped-capture"), + threadId, + }); + const resumed = yield* orchestrator.getThreadProjection(threadId); + assert.equal(resumed.runs.find((item) => item.id === queued.id)?.status, "starting"); + assert.notEqual(resumed.runs.find((item) => item.id === queued.id)?.queueHeld, true); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index aa44a6171e557..b02883c0105ba 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -201,6 +201,7 @@ const layerCheckpointCaptureServiceProvided = CheckpointCaptureService.layer.pip layerCheckpointServiceProvided, layerEventSinkProvided, IdAllocator.layer, + ThreadCommandExecutor.layer, ProjectionStore.layer, ), ), diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 0342a3ccc9d94..faf330058ea02 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -434,6 +434,7 @@ export function layerWithRegistry( layerCheckpointServiceProvided, layerEventSinkProvided, IdAllocator.layer, + ThreadCommandExecutor.layer, layerStores, ), ),