From bf5e2e3b2c3117f782f7f8f2f40830f4f0635d70 Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Fri, 25 Sep 2026 04:49:20 -0400 Subject: [PATCH] fix(orchestrator): Recover cancelled deliveries only on explicit rearm --- .../DelegatedCompletionDelivery.test.ts | 1084 ++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 145 ++- 2 files changed, 1160 insertions(+), 69 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 739d6983fd49..3caf3a141de7 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -19,13 +19,15 @@ import { ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; -import * as Effect from "effect/Effect"; import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; +import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; import * as Tracer from "effect/Tracer"; +import { SqlClient } from "effect/sql"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as ServerConfig from "../config.ts"; @@ -43,8 +45,8 @@ import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; -import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderContinuationService from "./ProviderContinuationService.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; @@ -57,7 +59,6 @@ import { continueRestartedRun } from "./RestartContinuation.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; -import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; const layerPlatformTest = Layer.merge( NodeServices.layer, @@ -120,47 +121,217 @@ const layerTestProviderInstanceRegistry = Layer.succeed( }, ); -const layerTest = Layer.mergeAll( - RuntimeLayer.layer, - RuntimeLayer.layerEventSink, - ProjectionStore.layer, - IdAllocator.layer, - ProviderContinuationRequests.layer, -).pipe( - Layer.provideMerge(RuntimeLayer.layerProjectService), - Layer.provide( - Layer.mock(WorkspacePaths.WorkspacePaths)({ - normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot), - }), - ), - Layer.provide(ProviderTurnStartServiceTestkit.layer), - Layer.provide( - Layer.succeed(ProjectEnrichmentService.ProjectEnrichmentService, { - peek: () => - Effect.succeed({ - repositoryIdentity: null, - faviconPath: null, - repositoryIdentityResolved: false, - }), - request: () => Effect.void, - getAvailable: () => - Effect.succeed({ - repositoryIdentity: null, - faviconPath: null, - repositoryIdentityResolved: false, +const layerTestWithDatabase = (database: Layer.Layer) => + Layer.mergeAll( + RuntimeLayer.layer, + RuntimeLayer.layerEventSink, + ProjectionStore.layer, + IdAllocator.layer, + ProviderContinuationRequests.layer, + ).pipe( + Layer.provideMerge(RuntimeLayer.layerProjectService), + Layer.provide( + Layer.mock(WorkspacePaths.WorkspacePaths)({ + normalizeWorkspaceRoot: (workspaceRoot) => Effect.succeed(workspaceRoot), + }), + ), + Layer.provide(ProviderTurnStartServiceTestkit.layer), + Layer.provide( + Layer.succeed(ProjectEnrichmentService.ProjectEnrichmentService, { + peek: () => + Effect.succeed({ + repositoryIdentity: null, + faviconPath: null, + repositoryIdentityResolved: false, + }), + request: () => Effect.void, + getAvailable: () => + Effect.succeed({ + repositoryIdentity: null, + faviconPath: null, + repositoryIdentityResolved: false, + }), + invalidate: () => Effect.void, + subscribeChanges: Effect.never, + }), + ), + Layer.provide(McpSessionRegistryTestkit.layer), + Layer.provide(McpProviderSessions.layer), + Layer.provideMerge(database), + Layer.provide(layerCheckpointStoreTest), + Layer.provide(layerServerConfig), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerTestProviderInstanceRegistry), + Layer.provide(layerPlatformTest), + ); + +const layerTest = layerTestWithDatabase(SqlitePersistence.layerMemory); + +it.effect("live upgrade reserves from the planned state", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("planned-state"); + const runId = RunId.make("planned-state:parent"); + const xId = NodeId.make("planned-state:x"); + const yId = NodeId.make("planned-state:y"); + yield* seedParentWithTerminalTask({ + threadId, + runId, + taskId: NodeId.make("planned-state:control"), + now, + projectId: ProjectId.make("planned-state:project"), + rootNodeId: NodeId.make("planned-state:root"), + completionWake: "always", + deliveryState: "delivered", + parentStatus: "running", + }); + const seed = yield* orchestrator.getThreadProjection(threadId); + const { completionDelivery: _omit, ...base } = seed.subagents[0]!; + yield* sink.write({ + events: [ + { + id: EventId.make("planned-state:x"), + type: "subagent.updated", + threadId, + runId, + occurredAt: now, + payload: { ...base, id: xId, completionWake: "settled_only" }, + }, + { + id: EventId.make("planned-state:y"), + type: "subagent.updated", + threadId, + runId, + occurredAt: now, + payload: { + ...base, + id: yId, + completionWake: "always", + completionDelivery: { state: "pending", observedByRunId: null }, + }, + }, + ], + }); + const upgrade = yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("planned-state:upgrade"), + parentThreadId: threadId, + taskId: xId, + completionWake: "always", + }); + const reservations = upgrade.storedEvents.flatMap(({ event }) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.delegatedCompletion?.delivery != null + ? [event.payload.delegatedCompletion.delivery] + : [], + ); + const after = yield* orchestrator.getThreadProjection(threadId); + const cohort = after.runs.find((run) => run.id === runId)?.delegatedCompletion; + assert.equal(reservations.length, 1, "one reservation per generation"); + assert.deepEqual(cohort?.delivery?.taskIds, [xId]); + assert.equal(cohort?.delivery?.generation, 2); + assert.equal(cohort?.nextGeneration, 3); + assert.equal( + after.subagents.find((task) => task.id === xId)?.completionDelivery?.state, + "claimed", + ); + assert.equal( + after.subagents.find((task) => task.id === yId)?.completionDelivery?.state, + "pending", + ); + }).pipe(Effect.provide(Layer.fresh(layerTest))), +); + +it.effect("rearmed target keeps its own new policy", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("target-policy"); + const runId = RunId.make("target-policy:parent"); + const taskId = NodeId.make("target-policy:task"); + yield* seedParentWithTerminalTask({ + threadId, + runId, + taskId, + now, + projectId: ProjectId.make("target-policy:project"), + rootNodeId: NodeId.make("target-policy:root"), + completionWake: "always", + deliveryState: "pending", + parentStatus: "completed", + }); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("target-policy:rearm"), + parentThreadId: threadId, + taskId, + completionWake: "settled_only", + }); + const after = yield* orchestrator.getThreadProjection(threadId); + const task = after.subagents.find((candidate) => candidate.id === taskId); + assert.equal(task?.completionDelivery?.state, "claimed"); + assert.equal( + task?.completionWake, + "settled_only", + "the command's own policy change was clobbered", + ); + }).pipe(Effect.provide(Layer.fresh(layerTest))), +); + +it.effect.each(["running", "completed"] as const)( + "reservation planning failures retain the wake-policy command identity (%s)", + (parentStatus) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`planning-error:${parentStatus}`); + const taskId = NodeId.make("planning-error:task"); + yield* seedParentWithTerminalTask({ + threadId, + runId: RunId.make("planning-error:parent"), + taskId, + now, + projectId: ProjectId.make("planning-error:project"), + rootNodeId: NodeId.make("planning-error:root"), + completionWake: "settled_only", + deliveryState: "pending", + parentStatus, + }); + const command = { + type: "delegated_task.wake-policy" as const, + commandId: CommandId.make("planning-error:upgrade"), + parentThreadId: threadId, + taskId, + completionWake: "always" as const, + }; + const original = store.getMessageCount; + yield* Effect.acquireUseRelease( + Effect.sync(() => { + Object.assign(store, { + getMessageCount: () => + Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId })), + }); }), - invalidate: () => Effect.void, - subscribeChanges: Effect.never, - }), - ), - Layer.provide(McpSessionRegistryTestkit.layer), - Layer.provide(McpProviderSessions.layer), - Layer.provide(SqlitePersistence.layerMemory), - Layer.provide(layerCheckpointStoreTest), - Layer.provide(layerServerConfig), - Layer.provide(ServerSettings.layerTest()), - Layer.provide(layerTestProviderInstanceRegistry), - Layer.provide(layerPlatformTest), + () => + Effect.gen(function* () { + const error = yield* orchestrator.dispatch(command).pipe(Effect.flip); + assert.equal(error._tag, "OrchestratorDispatchError"); + if (error._tag !== "OrchestratorDispatchError") return; + assert.equal(error.commandId, command.commandId); + assert.equal(error.commandType, command.type); + }), + () => + Effect.sync(() => { + Object.assign(store, { getMessageCount: original }); + }), + ); + const retry = yield* orchestrator.dispatch(command).pipe(Effect.flip); + assert.equal(retry._tag, "OrchestratorCommandPreviouslyRejectedError"); + }).pipe(Effect.provide(Layer.fresh(layerTest))), ); const seedParentWithTerminalTask = (input: { @@ -169,8 +340,9 @@ const seedParentWithTerminalTask = (input: { readonly runId: RunId; readonly rootNodeId: NodeId; readonly taskId: NodeId; - readonly deliveryState: "delivered" | "claimed" | "acknowledged" | "disposed"; + readonly deliveryState: "pending" | "delivered" | "claimed" | "acknowledged" | "disposed"; readonly completionWake?: "always" | "settled_only"; + readonly parentStatus?: "running" | "completed"; readonly deliveryTaskIds?: ReadonlyArray; readonly now: DateTime.Utc; }) => @@ -254,10 +426,10 @@ const seedParentWithTerminalTask = (input: { userMessageId: MessageId.make(`message:seed-user:${input.threadId}`), rootNodeId: input.rootNodeId, activeAttemptId: null, - status: "running", + status: input.parentStatus ?? "running", requestedAt: input.now, startedAt: input.now, - completedAt: null, + completedAt: input.parentStatus === "completed" ? input.now : null, checkpointId: null, contextHandoffId: null, delegatedCompletion: { @@ -384,6 +556,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { messageId, }); const accepted = yield* orchestrator.getThreadProjection(threadId); + assert.notEqual(accepted.runs.find((run) => run.id === runId)?.userMessageId, messageId); assert.equal( accepted.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, "delivered", @@ -455,7 +628,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { payload: { id: messageId, threadId, - runId, + runId: RunId.make("mailbox-mixed-continuation"), nodeId: task.parentNodeId, role: "user", text: "Background task finished", @@ -767,6 +940,823 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); }), ); + + it.effect("reserves concurrent explicit rearms atomically after a failed reservation write", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const threadId = ThreadId.make("atomic-rearm"); + const runId = RunId.make("atomic-rearm:parent"); + const deliveredId = NodeId.make("atomic-rearm:delivered"); + const siblingIds = [NodeId.make("atomic-rearm:third"), NodeId.make("atomic-rearm:second")]; + yield* seedParentWithTerminalTask({ + threadId, + runId, + taskId: deliveredId, + now, + projectId: ProjectId.make("atomic-rearm:project"), + rootNodeId: NodeId.make("atomic-rearm:root"), + completionWake: "always", + deliveryState: "delivered", + parentStatus: "completed", + }); + const seed = yield* orchestrator.getThreadProjection(threadId); + yield* sink.write({ + events: [ + ...siblingIds, + ...["one", "two", "three"].map((name) => NodeId.make(`atomic-rearm:control:${name}`)), + ].map((id) => ({ + id: EventId.make(`seed:${id}`), + type: "subagent.updated" as const, + threadId, + runId, + occurredAt: now, + payload: { + ...seed.subagents[0]!, + id, + completionDelivery: { + state: siblingIds.includes(id) ? ("pending" as const) : ("delivered" as const), + observedByRunId: null, + }, + }, + })), + }); + const rearm = (id: string, taskId = deliveredId) => + orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make(id), + parentThreadId: threadId, + taskId, + completionWake: "settled_only", + }); + yield* sql.unsafe( + "CREATE TEMP TRIGGER fail_reservation BEFORE UPDATE ON orchestration_v2_projection_runs WHEN NEW.run_id = 'atomic-rearm:parent' BEGIN SELECT RAISE(ABORT, 'simulated reservation write failure'); END", + ); + const beforeSequence = yield* sink.latestSequence(); + yield* rearm("atomic-rearm:failed").pipe(Effect.flip); + assert.equal(yield* sink.latestSequence(), beforeSequence); + const failed = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + failed.subagents.find((task) => task.id === deliveredId)?.completionWake, + "always", + ); + assert.equal(failed.runs[0]?.delegatedCompletion?.delivery, null); + assert.equal(failed.runs[0]?.delegatedCompletion?.nextGeneration, 2); + for (const id of siblingIds) + assert.equal( + failed.subagents.find((task) => task.id === id)?.completionDelivery?.state, + "pending", + ); + yield* sql.unsafe("DROP TRIGGER fail_reservation"); + const afterSequence = yield* sink.latestSequence(); + // Retry the same command after rollback, not an unrelated policy toggle. + yield* rearm("atomic-rearm:failed"); + yield* Effect.all( + ["one", "two", "three"].map((name) => + rearm(`atomic-rearm:${name}`, NodeId.make(`atomic-rearm:control:${name}`)), + ), + { concurrency: "unbounded" }, + ); + const reserved = yield* orchestrator.getThreadProjection(threadId); + const cohort = reserved.runs[0]!.delegatedCompletion!; + assert.equal(cohort.nextGeneration, 3); + assert.deepEqual(cohort.delivery?.taskIds, siblingIds.toSorted()); + assert.equal( + reserved.subagents.find((task) => task.id === deliveredId)?.completionDelivery?.state, + "delivered", + ); + for (const id of siblingIds) + assert.equal( + reserved.subagents.find((task) => task.id === id)?.completionDelivery?.state, + "claimed", + ); + const throughSequence = yield* sink.latestSequence(); + const events = yield* sink.stream({ afterSequence }).pipe( + Stream.takeUntil((stored) => stored.sequence >= throughSequence), + Stream.runCollect, + ); + assert.equal( + events.filter( + (stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === runId && + stored.event.payload.delegatedCompletion?.delivery != null, + ).length, + 1, + ); + const rebuilt = yield* Orchestrator.OrchestratorV2.pipe( + Effect.provide(Layer.fresh(layerTestWithDatabase(Layer.succeed(SqlClient.SqlClient, sql)))), + ); + yield* rebuilt.recoverDelegatedTasks; + const recovered = yield* rebuilt.getThreadProjection(threadId); + assert.deepEqual(recovered.runs[0]?.delegatedCompletion, cohort); + for (const id of siblingIds) + assert.equal( + recovered.subagents.find((task) => task.id === id)?.completionDelivery?.state, + "claimed", + ); + }).pipe(Effect.provide(Layer.fresh(layerTest))), + ); + + it.effect.each([ + "policy change", + "recovery", + "accept before finalization", + "accept interrupted before finalization", + "settlement", + "user cancellation", + ] as const)("handles cancelled delivery ownership via %s", (trigger) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:delegated-delivery-cancel-replan:${trigger}`); + const projectId = ProjectId.make("project:delegated-delivery-cancel-replan"); + const parentRunId = RunId.make("run:delegated-delivery-cancel-replan:parent"); + const deliveryRunId = RunId.make("run:delegated-delivery-cancel-replan:delivery"); + const parentRootNodeId = NodeId.make("node:delegated-delivery-cancel-replan:parent-root"); + const deliveryRootNodeId = NodeId.make("node:delegated-delivery-cancel-replan:delivery-root"); + const taskId = NodeId.make("node:delegated-delivery-cancel-replan:task"); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId: parentRunId, + rootNodeId: parentRootNodeId, + taskId, + deliveryState: "claimed", + completionWake: "always", + deliveryTaskIds: [taskId], + parentStatus: "completed", + now, + }); + const finalizerEntered = yield* Deferred.make(); + const releaseFinalizer = yield* Deferred.make(); + const store = yield* ProjectionStore.ProjectionStoreV2; + const getThread = store.getThread; + const acceptBeforeFinalization = + trigger === "accept before finalization" || + trigger === "accept interrupted before finalization"; + const terminalStatus = + trigger === "accept interrupted before finalization" ? "interrupted" : "cancelled"; + const queuedCancellation = trigger === "settlement" || trigger === "user cancellation"; + if (acceptBeforeFinalization) { + let held = false; + Object.assign(store, { + getThread: (id: ThreadId) => + Effect.gen(function* () { + const thread = yield* getThread(id); + if (id === threadId && !held) { + const state = yield* store.getThreadProjection(id); + if ( + state.runs.some( + (run) => run.id === deliveryRunId && run.status === terminalStatus, + ) + ) { + held = true; + yield* Deferred.succeed(finalizerEntered, undefined); + yield* Deferred.await(releaseFinalizer); + } + } + return thread; + }), + }); + } + const beforeCancel = yield* eventSink.latestSequence(); + yield* eventSink.write({ + commandId: CommandId.make("command:delegated-delivery-cancel-replan"), + events: [ + { + id: EventId.make("event:delegated-delivery-cancel-replan:message"), + type: "message.updated", + threadId, + runId: deliveryRunId, + nodeId: deliveryRootNodeId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + createdBy: "agent", + creationSource: "server", + id: messageId, + threadId, + runId: deliveryRunId, + nodeId: deliveryRootNodeId, + role: "user", + text: `Delegated task ${taskId} reached a terminal state.`, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + delegatedCompletion: { + parentRunId, + generation: 1, + taskIds: [taskId], + }, + }, + }, + { + id: EventId.make("event:delegated-delivery-cancel-replan:run"), + type: "run.updated", + threadId, + runId: deliveryRunId, + nodeId: deliveryRootNodeId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + id: deliveryRunId, + threadId, + ordinal: 2, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: ProviderThreadId.make( + `provider-thread:${String(threadId).replace("thread:", "")}`, + ), + userMessageId: messageId, + rootNodeId: deliveryRootNodeId, + activeAttemptId: null, + status: queuedCancellation ? "queued" : terminalStatus, + queueHeld: queuedCancellation, + requestedAt: now, + startedAt: queuedCancellation ? null : now, + completedAt: queuedCancellation ? null : now, + checkpointId: null, + contextHandoffId: null, + }, + }, + ], + }); + + if (queuedCancellation) { + if (trigger === "user cancellation") { + yield* orchestrator.dispatch({ + type: "queued-run.cancel", + commandId: CommandId.make("cancel-replan:user-cancel"), + threadId, + runId: deliveryRunId, + }); + } else { + yield* orchestrator.dispatch({ + type: "thread.settle", + commandId: CommandId.make("cancel-replan:settle"), + threadId, + }); + } + const settled = yield* orchestrator.getThreadProjection(threadId); + assert.equal(settled.runs.find((run) => run.id === deliveryRunId)?.status, "cancelled"); + assert.equal( + settled.subagents.find((task) => task.id === taskId)?.completionDelivery?.state, + "disposed", + ); + assert.equal( + settled.runs.find((run) => run.id === parentRunId)?.delegatedCompletion?.disposition, + "disposed", + ); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("cancel-replan:policy-after-settle"), + parentThreadId: threadId, + taskId, + completionWake: "settled_only", + }); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("cancel-replan:automatic-upgrade"), + parentThreadId: threadId, + taskId, + completionWake: "always", + }); + const afterPolicy = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterPolicy.runs.find((run) => run.id === parentRunId)?.delegatedCompletion?.delivery, + null, + ); + assert.equal( + afterPolicy.subagents.find((task) => task.id === taskId)?.completionDelivery?.state, + "disposed", + ); + return; + } + if (acceptBeforeFinalization) { + yield* Effect.gen(function* () { + yield* Deferred.await(finalizerEntered); + const durable = yield* store.getThreadProjection(threadId); + assert.equal( + durable.runs.find((run) => run.id === deliveryRunId)?.status, + terminalStatus, + ); + assert.equal( + durable.runs.find((run) => run.id === parentRunId)?.delegatedCompletion?.delivery + ?.messageId, + messageId, + ); + const task = durable.subagents.find((row) => row.id === taskId)!; + const siblingId = NodeId.make("cancel-pre-finalize:pending-sibling"); + yield* eventSink.write({ + events: [ + { + id: EventId.make("cancel-pre-finalize:pending-sibling"), + type: "subagent.updated", + threadId, + runId: parentRunId, + occurredAt: now, + payload: { + ...task, + id: siblingId, + completionDelivery: { state: "pending", observedByRunId: null }, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "notification.delivery.accept", + commandId: CommandId.make("cancel-pre-finalize:accept"), + threadId, + messageId, + }); + const after = yield* store.getThreadProjection(threadId); + assert.equal( + after.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + "claimed", + ); + assert.equal( + after.subagents.find((row) => row.id === siblingId)?.completionDelivery?.state, + "pending", + ); + assert.deepEqual( + after.runs.find((run) => run.id === parentRunId)?.delegatedCompletion, + durable.runs.find((run) => run.id === parentRunId)?.delegatedCompletion, + ); + }).pipe( + Effect.ensuring(Deferred.succeed(releaseFinalizer, undefined)), + Effect.ensuring(Effect.sync(() => Object.assign(store, { getThread }))), + ); + } + if (acceptBeforeFinalization) return; + // Wait on the durable cohort-finalization event, not scheduler timing. + yield* eventSink.stream({ afterSequence: beforeCancel, eventType: "run.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && stored.event.payload.id === parentRunId, + ), + Stream.take(1), + Stream.runDrain, + ); + const afterCancel = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterCancel.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery + ?.state, + "pending", + ); + assert.equal( + afterCancel.runs.find((candidate) => candidate.id === parentRunId)?.delegatedCompletion + ?.delivery, + null, + ); + + yield* orchestrator.dispatch({ + type: "notification.delivery.accept", + commandId: CommandId.make("command:cancel-replan:late-acceptance"), + threadId, + messageId, + }); + const afterLateAcceptance = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterLateAcceptance.runs.find((run) => run.id === parentRunId)?.delegatedCompletion + ?.delivery, + null, + ); + assert.equal( + afterLateAcceptance.subagents.find((task) => task.id === taskId)?.completionDelivery?.state, + "pending", + ); + assert.notInclude(yield* store.getRecoveryThreadIds("delegated-completions"), threadId); + if (trigger === "policy change") { + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("command:delegated-delivery-cancel-replan:wake-policy"), + parentThreadId: threadId, + taskId, + completionWake: "settled_only", + }); + } else { + // Rebuild over the same database: cancellation must survive ordinary restart. + const sql = yield* SqlClient.SqlClient; + const otherRunId = RunId.make("cancel-replan:other-parent"); + const otherTaskId = NodeId.make("cancel-replan:other-task"); + const cancelledProjection = yield* orchestrator.getThreadProjection(threadId); + yield* eventSink.write({ + events: [ + { + id: EventId.make("cancel-replan:other-run"), + type: "run.updated", + threadId, + runId: otherRunId, + occurredAt: now, + payload: { + ...cancelledProjection.runs.find((run) => run.id === parentRunId)!, + id: otherRunId, + ordinal: 3, + delegatedCompletion: { + disposition: "open", + nextGeneration: 2, + delivery: { + generation: 1, + messageId: MessageId.make("cancel-replan:other-message"), + taskIds: [otherTaskId], + }, + }, + }, + }, + { + id: EventId.make("cancel-replan:other-task"), + type: "subagent.updated", + threadId, + runId: otherRunId, + occurredAt: now, + payload: { + ...cancelledProjection.subagents.find((task) => task.id === taskId)!, + id: otherTaskId, + runId: otherRunId, + completionDelivery: { state: "claimed", observedByRunId: null }, + }, + }, + ], + }); + assert.include(yield* store.getRecoveryThreadIds("delegated-completions"), threadId); + yield* Orchestrator.OrchestratorV2.pipe( + Effect.flatMap((rebuilt) => rebuilt.recoverDelegatedTasks), + Effect.provide( + Layer.fresh(layerTestWithDatabase(Layer.succeed(SqlClient.SqlClient, sql))), + ), + ); + const afterRestart = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterRestart.runs.find((run) => run.id === parentRunId)?.delegatedCompletion?.delivery, + null, + ); + assert.equal( + afterRestart.subagents.find((task) => task.id === taskId)?.completionDelivery?.state, + "pending", + ); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("cancel-replan:explicit-after-restart"), + parentThreadId: threadId, + taskId, + completionWake: "settled_only", + }); + } + + const afterReplan = yield* orchestrator.getThreadProjection(threadId); + const parentRun = afterReplan.runs.find((candidate) => candidate.id === parentRunId); + const task = afterReplan.subagents.find((candidate) => candidate.id === taskId); + assert.equal(task?.completionDelivery?.state, "claimed"); + assert.isNotNull(parentRun?.delegatedCompletion?.delivery ?? null); + assert.deepEqual(parentRun?.delegatedCompletion?.delivery?.taskIds, [taskId]); + }).pipe(Effect.provide(Layer.fresh(layerTest))), + ); + it.effect("does not automatically re-reserve a cancelled delivery", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:delegated-delivery-cancel-replan:standalone`); + const projectId = ProjectId.make("project:delegated-delivery-cancel-replan"); + const parentRunId = RunId.make("run:delegated-delivery-cancel-replan:parent"); + const deliveryRunId = RunId.make("run:delegated-delivery-cancel-replan:delivery"); + const parentRootNodeId = NodeId.make("node:delegated-delivery-cancel-replan:parent-root"); + const deliveryRootNodeId = NodeId.make("node:delegated-delivery-cancel-replan:delivery-root"); + const taskId = NodeId.make("node:delegated-delivery-cancel-replan:task"); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId: parentRunId, + rootNodeId: parentRootNodeId, + taskId, + deliveryState: "claimed", + completionWake: "always", + deliveryTaskIds: [taskId], + parentStatus: "completed", + now, + }); + const beforeCancel = yield* eventSink.latestSequence(); + yield* eventSink.write({ + commandId: CommandId.make("command:delegated-delivery-cancel-replan"), + events: [ + { + id: EventId.make("event:delegated-delivery-cancel-replan:message"), + type: "message.updated", + threadId, + runId: deliveryRunId, + nodeId: deliveryRootNodeId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + createdBy: "agent", + creationSource: "server", + id: messageId, + threadId, + runId: deliveryRunId, + nodeId: deliveryRootNodeId, + role: "user", + text: `Delegated task ${taskId} reached a terminal state.`, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + delegatedCompletion: { + parentRunId, + generation: 1, + taskIds: [taskId], + }, + }, + }, + { + id: EventId.make("event:delegated-delivery-cancel-replan:run"), + type: "run.updated", + threadId, + runId: deliveryRunId, + nodeId: deliveryRootNodeId, + providerInstanceId: modelSelection.instanceId, + occurredAt: now, + payload: { + id: deliveryRunId, + threadId, + ordinal: 2, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: ProviderThreadId.make( + `provider-thread:${String(threadId).replace("thread:", "")}`, + ), + userMessageId: messageId, + rootNodeId: deliveryRootNodeId, + activeAttemptId: null, + status: "cancelled", + queueHeld: false, + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + ], + }); + + // Wait on the durable cohort-finalization event, not scheduler timing. + yield* eventSink.stream({ afterSequence: beforeCancel, eventType: "run.updated" }).pipe( + Stream.filter( + (stored) => + stored.event.type === "run.updated" && stored.event.payload.id === parentRunId, + ), + Stream.take(1), + Stream.runDrain, + ); + const afterCancel = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterCancel.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery + ?.state, + "pending", + ); + assert.equal( + afterCancel.runs.find((candidate) => candidate.id === parentRunId)?.delegatedCompletion + ?.delivery, + null, + ); + + // Startup recovery does not re-reserve a cancellation the live listener already settled. + yield* orchestrator.recoverDelegatedTasks; + const afterRecovery = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + afterRecovery.subagents.find((candidate) => candidate.id === taskId)?.completionDelivery + ?.state, + "pending", + ); + assert.equal( + afterRecovery.runs.find((candidate) => candidate.id === parentRunId)?.delegatedCompletion + ?.delivery, + null, + ); + }).pipe(Effect.provide(Layer.fresh(layerTest))), + ); + + it.effect("overlapping explicit rearms keep policy and reservation ownership", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("review-lock-probe"); + const runId = RunId.make("review-lock-probe:parent"); + const deliveredId = NodeId.make("review-lock-probe:delivered"); + const siblingX = NodeId.make("review-lock-probe:x"); + const siblingY = NodeId.make("review-lock-probe:y"); + const controlId = NodeId.make("review-lock-probe:control"); + yield* seedParentWithTerminalTask({ + threadId, + runId, + taskId: deliveredId, + now, + projectId: ProjectId.make("review-lock-probe:project"), + rootNodeId: NodeId.make("review-lock-probe:root"), + completionWake: "always", + deliveryState: "delivered", + parentStatus: "completed", + }); + const seed = yield* orchestrator.getThreadProjection(threadId); + yield* sink.write({ + events: [siblingX, siblingY, controlId].map((id) => ({ + id: EventId.make(`review-lock-probe:seed:${id}`), + type: "subagent.updated" as const, + threadId, + runId, + occurredAt: now, + payload: { + ...seed.subagents[0]!, + id, + completionWake: "always" as const, + completionDelivery: { + state: id === controlId ? ("delivered" as const) : ("pending" as const), + observedByRunId: null, + }, + }, + })), + }); + // Pause after the first snapshot, before reservation planning finishes. + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const getMessageCount = store.getMessageCount; + let calls = 0; + Object.assign(store, { + getMessageCount: (id: ThreadId) => + Effect.gen(function* () { + if (id === threadId && calls++ === 0) { + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + } + return yield* getMessageCount(id); + }), + }); + const afterSequence = yield* sink.latestSequence(); + const first = yield* orchestrator + .dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("review-lock-probe:first"), + parentThreadId: threadId, + taskId: controlId, + completionWake: "settled_only", + }) + .pipe(Effect.forkChild); + yield* Deferred.await(entered); + const second = yield* orchestrator + .dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make("review-lock-probe:second"), + parentThreadId: threadId, + taskId: siblingX, + completionWake: "settled_only", + }) + .pipe(Effect.forkChild); + yield* TestClock.adjust(0); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(first); + yield* Fiber.join(second); + Object.assign(store, { getMessageCount }); + const throughSequence = yield* sink.latestSequence(); + const events = yield* sink.stream({ afterSequence }).pipe( + Stream.takeUntil((stored) => stored.sequence >= throughSequence), + Stream.runCollect, + ); + const reservations = Array.from(events).flatMap((stored) => + stored.event.type === "run.updated" && + stored.event.payload.id === runId && + stored.event.payload.delegatedCompletion?.delivery != null + ? [stored.event.payload.delegatedCompletion.delivery] + : [], + ); + const after = yield* orchestrator.getThreadProjection(threadId); + const x = after.subagents.find((task) => task.id === siblingX); + const cohort = after.runs.find((run) => run.id === runId)?.delegatedCompletion; + assert.equal(x?.completionWake, "settled_only", "second rearm's committed policy was lost"); + assert.equal(reservations.length, 1, "more than one reservation was written"); + assert.deepEqual(cohort?.delivery?.taskIds, [siblingX, siblingY].toSorted()); + }).pipe(Effect.provide(Layer.fresh(layerTest))), + ); + + it.effect.each(["in-flight delivery", "live parent", "disposed cohort"] as const)( + "explicit rearm preserves the %s", + (guard) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`rearm-guard:${guard}`); + const runId = RunId.make(`${threadId}:parent`); + const taskId = NodeId.make(`${threadId}:claimed`); + const siblingId = NodeId.make(`${threadId}:pending`); + const controlId = NodeId.make(`${threadId}:control`); + yield* seedParentWithTerminalTask({ + threadId, + runId, + taskId, + now, + projectId: ProjectId.make(`${threadId}:project`), + rootNodeId: NodeId.make(`${threadId}:root`), + parentStatus: guard === "live parent" ? "running" : "completed", + completionWake: "always", + deliveryState: guard === "in-flight delivery" ? "claimed" : "delivered", + ...(guard === "in-flight delivery" ? { deliveryTaskIds: [taskId] } : {}), + }); + const seed = yield* orchestrator.getThreadProjection(threadId); + const deliveryRun = { ...seed.runs[0]! }; + delete deliveryRun.delegatedCompletion; + yield* sink.write({ + events: [ + ...(guard === "in-flight delivery" + ? [ + { + id: EventId.make(`${threadId}:in-flight`), + type: "run.updated" as const, + threadId, + runId: RunId.make(`${threadId}:delivery`), + occurredAt: now, + payload: { + ...deliveryRun, + id: RunId.make(`${threadId}:delivery`), + ordinal: 2, + status: "running" as const, + userMessageId: seed.runs[0]!.delegatedCompletion!.delivery!.messageId, + }, + }, + ] + : []), + ...[siblingId, controlId].map((id) => ({ + id: EventId.make(`seed:${id}`), + type: "subagent.updated" as const, + threadId, + runId, + occurredAt: now, + payload: { + ...seed.subagents[0]!, + id, + completionWake: + id === siblingId && guard === "live parent" + ? ("settled_only" as const) + : ("always" as const), + completionDelivery: { + state: id === siblingId ? ("pending" as const) : ("delivered" as const), + observedByRunId: null, + }, + }, + })), + ...(guard === "disposed cohort" + ? [ + { + id: EventId.make(`${threadId}:dispose`), + type: "run.updated" as const, + threadId, + runId, + occurredAt: now, + payload: { + ...seed.runs[0]!, + delegatedCompletion: { + ...seed.runs[0]!.delegatedCompletion!, + disposition: "disposed" as const, + }, + }, + }, + ] + : []), + ], + }); + const before = yield* orchestrator.getThreadProjection(threadId); + yield* orchestrator.dispatch({ + type: "delegated_task.wake-policy", + commandId: CommandId.make(`${threadId}:rearm`), + parentThreadId: threadId, + taskId: controlId, + completionWake: "settled_only", + }); + const after = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + after.runs.find((run) => run.id === runId)?.delegatedCompletion, + before.runs.find((run) => run.id === runId)?.delegatedCompletion, + ); + assert.equal( + after.subagents.find((task) => task.id === siblingId)?.completionDelivery?.state, + "pending", + ); + assert.deepEqual( + after.subagents.find((task) => task.id === taskId)?.completionDelivery, + before.subagents.find((task) => task.id === taskId)?.completionDelivery, + ); + }).pipe(Effect.provide(Layer.fresh(layerTest))), + ); }); // Runtime reconciliation writes restart cancellations under this command diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 9b83a9abde2e..8d67b7de83f5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -867,13 +867,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); - const mapDelegatedCompletionError = (effect: Effect.Effect) => + const mapDelegatedCompletionError = ( + effect: Effect.Effect, + command?: Pick, + ) => effect.pipe( Effect.mapError( (cause) => new OrchestratorDispatchError({ - commandId: CommandId.make("command:system:delegated-completion-delivery"), - commandType: "delegated_task.completion-delivery", + commandId: + command?.commandId ?? CommandId.make("command:system:delegated-completion-delivery"), + commandType: command?.type ?? "delegated_task.completion-delivery", cause, }), ), @@ -1246,6 +1250,84 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); }); + const reservePendingDelegatedCompletionDelivery = Effect.fnUntraced(function* ( + projection: Pick< + OrchestrationV2ThreadProjection, + "thread" | "runs" | "subagents" | "providerTurns" + >, + parentRun: OrchestrationV2Run, + emitEvent: ReturnType, + command: OrchestrationV2Command, + ) { + const threadId = projection.thread.id; + const cohort = parentRun.delegatedCompletion; + if ( + cohort === undefined || + cohort.disposition !== "open" || + cohort.delivery !== null || + projection.thread.archivedAt !== null || + projection.thread.deletedAt !== null + ) + return; + const parentIsLive = hasLiveRun(projection); + const pendingTasks = projection.subagents.filter( + (task) => + task.origin === "app_owned" && + task.runId === parentRun.id && + isTerminalDelegatedTaskStatus(task.status) && + task.completionDelivery?.state === "pending" && + (!parentIsLive || task.completionWake === "always"), + ); + if (pendingTasks.length === 0) return; + const now = yield* DateTime.now; + const messageId = yield* mapDelegatedCompletionError( + idAllocator.allocate.message({ + threadId, + ordinal: + (yield* mapDelegatedCompletionError(projectionStore.getMessageCount(threadId), command)) + + 1, + }), + command, + ); + for (const event of [ + ...pendingTasks.map((task) => ({ + type: "subagent.updated" as const, + threadId, + ...(task.runId === null ? {} : { runId: task.runId }), + nodeId: task.id, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + occurredAt: now, + payload: { + ...task, + completionDelivery: { state: "claimed" as const, observedByRunId: null }, + updatedAt: now, + }, + })), + { + type: "run.updated" as const, + threadId, + runId: parentRun.id, + ...(parentRun.rootNodeId === null ? {} : { nodeId: parentRun.rootNodeId }), + providerInstanceId: parentRun.providerInstanceId, + occurredAt: now, + payload: { + ...parentRun, + delegatedCompletion: { + ...cohort, + nextGeneration: cohort.nextGeneration + 1, + delivery: { + generation: cohort.nextGeneration, + messageId, + taskIds: pendingTasks.map((task) => task.id), + }, + }, + }, + }, + ]) + yield* emitEvent(event); + }); + const offerDelegatedCompletionDeliveries = (threadId: ThreadId) => Effect.gen(function* () { const projection = yield* projectionStore.getThreadRecords(threadId, ["runs"]); @@ -7396,17 +7478,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio completionWake: command.completionWake, updatedAt: now, }; - // A non-terminal task needs no offer here: finalize reads the upgraded - // policy when the child terminalizes. Both writers hold this parent lock, - // so a terminal task means finalize already committed the terminal row and - // already made its offer decision under the pre-upgrade policy: under - // settled_only it offered iff the spawning run was not live. Plan a - // delivery only when that run is live now, which is precisely the case - // where finalize skipped. The mailbox steers a capable active session or - // queues behind that run. When that run is not live, finalize already - // offered and a second offer would wake the parent twice. (If the parent settled in between, - // this skips a wake that finalize also skipped; a missed wake is cheaper - // than a duplicate one, and the result is already in the projection.) + // Preserve live-parent batching, then reserve other pending cohorts from + // the resulting plan. Non-terminal tasks use the policy when they finish. + // A settled_only result finalized while the parent ran still gets no wake if the parent finished before this upgrade. const parentRun = task.runId === null ? undefined @@ -7422,6 +7496,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio task: updatedTask, updatedTask, now, + command, }) : undefined; yield* emitEvent({ @@ -7434,10 +7509,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: completionPlan?.task ?? updatedTask, }); - if (completionPlan === undefined) { - return; - } - if (completionPlan.parentRun !== undefined) { + if (completionPlan?.parentRun !== undefined) { yield* emitEvent({ type: "run.updated", threadId: command.parentThreadId, @@ -7450,7 +7522,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: completionPlan.parentRun, }); } - if (completionPlan.message !== undefined) { + if (completionPlan?.message !== undefined) { yield* emitEvent({ type: "message.updated", threadId: command.parentThreadId, @@ -7465,6 +7537,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: completionPlan.message, }); } + // Reserve from the planned policy state so policy, claims and generation commit together. + const pending = { + ...parentProjection, + subagents: parentProjection.subagents.map((candidate) => + candidate.id === task.id ? (completionPlan?.task ?? updatedTask) : candidate, + ), + runs: parentProjection.runs.map((candidate) => + candidate.id === completionPlan?.parentRun?.id ? completionPlan.parentRun : candidate, + ), + }; + for (const run of pending.runs) + yield* reservePendingDelegatedCompletionDelivery(pending, run, emitEvent, command); }); const dispatchCreatedThreadRecord = Effect.fn("orchestrationV2.dispatch.createdThreadRecord")( @@ -10277,6 +10361,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly task: OrchestrationV2Subagent; readonly updatedTask: OrchestrationV2Subagent; readonly now: DateTime.Utc; + readonly command?: Pick; }) { const taskDelivery = input.updatedTask.completionDelivery; // delivered ownership has already settled through a completed wake run. @@ -10433,8 +10518,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ordinal: (yield* mapDelegatedCompletionError( projectionStore.getMessageCount(input.parentRun.threadId), + input.command, )) + 1, }), + input.command, ); const nextCohort = { disposition: "open" as const, @@ -10816,7 +10903,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); - const finalizeDelegatedCompletionDelivery = (threadId: ThreadId, runId: RunId) => + const finalizeDelegatedCompletionDelivery = ( + threadId: ThreadId, + runId: RunId, + options?: { readonly restartRecovery?: boolean }, + ) => Effect.gen(function* () { // Ordinary turns do not own a delegated delivery. Read just their input // before loading the cohort state needed to settle an actual delivery. @@ -10888,6 +10979,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cohort.disposition === "open" && projection.thread.archivedAt === null && projection.thread.deletedAt === null && + // A live cancellation leaves results pending until an explicit rearm. + // Startup recovery finalizes deliveries the live listener skipped, + // chiefly ones restart reconciliation cut, so those are offered again. + (deliveryRun.status !== "cancelled" || options?.restartRecovery === true) && pendingTaskIds.length > 0; const nextDelivery = canReserveFollowUp ? { @@ -10965,6 +11060,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) { const projection = yield* getProjectionWithPendingEvents(command.threadId, events); const message = projection.messages.find((row) => row.id === command.messageId); + const deliveryRun = projection.runs.find((run) => run.id === message?.runId); const ownership = message?.delegatedCompletion; const parentRun = projection.runs.find((run) => run.id === ownership?.parentRunId); const cohort = parentRun?.delegatedCompletion; @@ -10977,7 +11073,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio cohort.disposition !== "open" || delivery == null || delivery.messageId !== command.messageId || - delivery.generation !== ownership?.generation + delivery.generation !== ownership?.generation || + deliveryRun === undefined || + deliveryRun.status === "cancelled" || + deliveryRun.status === "interrupted" ) { yield* emitEvent({ type: "thread.metadata-updated", @@ -11740,7 +11839,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ) .map((run) => run.id); for (const runId of terminalDeliveryRunIds) { - yield* finalizeDelegatedCompletionDelivery(threadId, runId); + yield* finalizeDelegatedCompletionDelivery(threadId, runId, { + restartRecovery: true, + }); } const refreshed = terminalDeliveryRunIds.length === 0