diff --git a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts new file mode 100644 index 000000000000..97a59171278b --- /dev/null +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -0,0 +1,544 @@ +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + ThreadId, + type OrchestrationV2DomainEvent, + type OrchestrationV2Subagent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Latch from "effect/Latch"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as EventSink from "./EventSink.ts"; +import * as Orchestrator from "./Orchestrator.ts"; +import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; +import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; + +const driver = ProviderDriverKind.make("codex"); +const instanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId, model: "test-model" }; +const taskA = NodeId.make("task:a"); +const taskB = NodeId.make("task:b"); +const capabilities = { + ...CodexProviderCapabilitiesV2, + turns: { ...CodexProviderCapabilitiesV2.turns, supportsActiveSteering: true }, +}; + +// A parent delegated two tasks with completion wakes. Child A's result was +// reserved as a steer into the parent's running turn; child B finished while +// that reservation was outstanding, so it waits behind it. The provider stops +// accepting steers just before its turn ends (seen with Pi), so the steer +// never lands while the parent still looks running. +it.effect.each([ + { + name: "unaccepted", + title: + "re-offers a steered delegated delivery whose steer never landed when the parent settles", + resultRead: false, + queuedMessage: false, + }, + { + name: "read", + title: "releases a steered delegated delivery the parent already read when it settles", + resultRead: true, + queuedMessage: false, + }, + { + // B waits only for settlement, and a queued user message starts the moment + // the parent settles. B must still wake the parent after that message. + name: "read-queued", + title: "reserves a settled-only sibling when a queued message starts at settlement", + resultRead: true, + queuedMessage: true, + }, + { + // The settlement offers A's wake, and the user rolls the parent turn back + // before the continuation worker dispatches it. + name: "rolled-back", + title: "drops a recovered delegated wake whose parent turn was rolled back before it started", + resultRead: false, + queuedMessage: false, + rollback: true, + }, +] as const)("$title", (scenario) => { + const { resultRead, queuedMessage } = scenario; + const rollback = "rollback" in scenario && scenario.rollback; + return Effect.scoped( + Effect.gen(function* () { + const name = `delegated-steer-settlement-${scenario.name}`; + const cwd = yield* checkpointWorkspace(name); + const events = yield* Queue.unbounded(); + const started: ProviderAdapter.ProviderAdapterV2TurnInput[] = []; + let steerCalls = 0; + // The continuation worker's queue. A closed gate holds the request the + // worker took, and `workerReady` signals each time the worker asks for + // its next request, so after a release it marks the dispatch finished. + const continuationQueue = + yield* Queue.unbounded(); + const continuationOffers = + yield* Queue.unbounded(); + const workerReady = yield* Queue.unbounded(); + const continuationGate = yield* Latch.make(!rollback); + const continuationRequests = Layer.succeed( + ProviderContinuationRequests.ProviderContinuationRequests, + { + offer: (request) => + Queue.offer(continuationOffers, request).pipe( + Effect.andThen(Queue.offer(continuationQueue, request)), + Effect.asVoid, + ), + take: Effect.gen(function* () { + yield* Queue.offer(workerReady, undefined); + const request = yield* Queue.take(continuationQueue); + yield* continuationGate.await; + return request; + }), + }, + ); + const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(capabilities), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + const now = yield* DateTime.now; + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready", + cwd, + model: modelSelection.model, + capabilities, + createdAt: now, + updatedAt: now, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: ({ threadId }) => + Effect.succeed({ + id: ProviderThreadId.make(`provider-thread:${threadId}`), + driver, + providerInstanceId: instanceId, + providerSessionId: input.providerSessionId, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: null, + lastRunOrdinal: null, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }), + resumeThread: ({ providerThread }) => Effect.succeed(providerThread), + startTurn: (turn) => + Effect.gen(function* () { + started.push(turn); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { + driver, + nativeId: `native:${turn.attemptId}`, + strength: "strong", + }, + ordinal: turn.providerTurnOrdinal, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + }), + steerTurn: (turn) => + Effect.gen(function* () { + steerCalls += 1; + return yield* new ProviderAdapter.ProviderAdapterSteerRunError({ + driver, + providerThreadId: turn.providerThread.id, + providerTurnId: turn.providerTurnId, + cause: "turn is not active", + }); + }), + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.void, + readThreadSnapshot: () => Effect.die("unused"), + rollbackThread: ({ providerThread }) => + Effect.succeed({ + providerThread, + providerTurns: [], + messages: [], + runtimeRequests: [], + }), + forkThread: () => Effect.die("unused"), + }; + }), + }; + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make(`thread:${name}`); + // The first matching event on the thread, already stored or still to + // come. The live-clock bound only turns a missing event into a failed + // assertion; a passing run never waits on it. + const awaitEvent = ( + types: ReadonlyArray, + predicate: ( + event: Extract, + ) => boolean, + ) => + sink.stream({ threadId }).pipe( + Stream.map((stored) => stored.event), + Stream.filter( + (event): event is Extract => + (types as ReadonlyArray).includes(event.type) && + predicate(event as Extract), + ), + Stream.runHead, + Effect.timeoutOption("10 seconds"), + TestClock.withLive, + Effect.map(Option.flatten), + ); + // startTurn's running turn is ingested on the adapter event fiber. + const runningTurn = (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + const running = yield* awaitEvent( + ["provider-turn.updated"], + (event) => + event.payload.runAttemptId === turn.attemptId && event.payload.status === "running", + ); + if (Option.isNone(running)) { + return assert.fail(`run ${turn.runId} has no running provider turn`); + } + return running.value.payload; + }); + // Ends the run's provider turn and waits for the run to settle. + const settle = (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + const turn = yield* runningTurn(input); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal: input.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + const runReached = (status: "waiting" | "completed") => + awaitEvent( + ["run.updated"], + (event) => event.payload.id === input.runId && event.payload.status === status, + ).pipe( + Effect.map((event) => + assert.isTrue(Option.isSome(event), `run ${input.runId} never ${status}`), + ), + ); + yield* runReached("waiting"); + yield* worker.drain(); + yield* runReached("completed"); + }); + // Waits for the run after the first `startedCount` turns to start, as + // the continuation worker or queue promotion does it, then runs its turn + // start. Queue promotion can already have started it while `settle` + // drained. When none starts, the assertions show the delivery + // reservation left behind. + const nextWake = ( + startedCount: number, + expected: { readonly taskIds: ReadonlyArray }, + ) => + Effect.gen(function* () { + const startedRunIds = new Set(started.slice(0, startedCount).map((turn) => turn.runId)); + // A wake is created starting; a promoted queued run is updated to it. + const starting = yield* awaitEvent( + ["run.created", "run.updated"], + (event) => + event.payload.status === "starting" && !startedRunIds.has(event.payload.id), + ); + if (Option.isNone(starting)) { + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.runs[0]?.delegatedCompletion?.delivery?.taskIds, + expected.taskIds, + "no wake started for the reserved delivery", + ); + return assert.fail("no wake started for the reserved delivery"); + } + yield* worker.drain(); + assert.equal(started.length, startedCount + 1); + const wake = started[startedCount]!; + assert.equal(wake.runId, starting.value.payload.id); + for (const taskId of [taskA, taskB]) { + if (expected.taskIds.includes(taskId)) { + assert.include(wake.message.text, String(taskId)); + } else { + assert.notInclude(wake.message.text, String(taskId)); + } + } + return wake; + }); + + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create"), + threadId, + projectId: ProjectId.make(`project:${name}`), + title: "Delegated steer settlement", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: cwd, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("first"), + threadId, + messageId: MessageId.make("message:first"), + text: "delegate two tasks", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + const parent = started[0]!; + yield* runningTurn(parent); + + const steerMessageId = MessageId.make("message:delegated-steer"); + const projection = yield* orchestrator.getThreadProjection(threadId); + const parentRun = projection.runs.find((run) => run.id === parent.runId)!; + const now = yield* DateTime.now; + const task = ( + id: NodeId, + completionDelivery: OrchestrationV2Subagent["completionDelivery"], + completionWake: OrchestrationV2Subagent["completionWake"], + ): OrchestrationV2Subagent => ({ + id, + threadId, + runId: parent.runId, + parentNodeId: parent.rootNodeId, + origin: "app_owned", + createdBy: "agent", + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: null, + prompt: `Do ${id}`, + title: null, + model: null, + completionWake, + completionDelivery, + status: "completed", + result: `${id} done`, + startedAt: now, + completedAt: now, + updatedAt: now, + }); + yield* sink.write({ + events: [ + { + id: EventId.make("seed:cohort"), + type: "run.updated", + threadId, + runId: parent.runId, + occurredAt: now, + payload: { + ...parentRun, + delegatedCompletion: { + disposition: "open", + nextGeneration: 2, + delivery: { generation: 1, messageId: steerMessageId, taskIds: [taskA] }, + }, + }, + }, + ...[ + task(taskA, { state: "claimed", observedByRunId: null }, "always"), + task( + taskB, + { state: "pending", observedByRunId: null }, + queuedMessage ? "settled_only" : "always", + ), + ].map((payload) => ({ + id: EventId.make(`seed:${payload.id}`), + type: "subagent.updated" as const, + threadId, + runId: parent.runId, + nodeId: payload.id, + occurredAt: now, + payload, + })), + ], + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("delegated-steer"), + threadId, + messageId: steerMessageId, + text: "Delegated task finished", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { parentRunId: parent.runId, generation: 1, taskIds: [taskA] }, + }); + // Every retry is rejected while the parent turn still looks running. + for (let attempt = 0; attempt < 5; attempt++) { + yield* worker.drain(); + yield* TestClock.adjust("1 minute"); + } + yield* worker.drain(); + assert.equal(steerCalls, 5); + if (resultRead) { + // task_status read A's result during the turn, emptying A's reservation. + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("read-a"), + parentThreadId: threadId, + taskId: taskA, + observedByRunId: parent.runId, + }); + } + const queuedMessageId = MessageId.make("message:queued"); + if (queuedMessage) { + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("queued"), + threadId, + messageId: queuedMessageId, + text: "one more thing", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + } + const beforeSettle = yield* orchestrator.getThreadProjection(threadId); + assert.equal(beforeSettle.runs[0]?.status, "running"); + assert.equal( + beforeSettle.runs[0]?.delegatedCompletion?.delivery?.messageId, + steerMessageId, + ); + + yield* settle(parent); + if (rollback) { + // The settlement offered A's wake; the worker holds it at the gate. + const offer = yield* Queue.take(continuationOffers).pipe( + Effect.timeoutOption("10 seconds"), + TestClock.withLive, + ); + assert.isTrue(Option.isSome(offer), "settlement offered no wake"); + assert.equal( + Option.getOrUndefined(offer)?.delegatedCompletion?.messageId, + steerMessageId, + ); + const settled = yield* orchestrator.getThreadProjection(threadId); + const threadStart = settled.checkpoints.find( + (checkpoint) => checkpoint.ordinalWithinScope === 0 && checkpoint.status === "ready", + ); + assert.isDefined(threadStart, "no thread-start checkpoint to roll back to"); + if (threadStart === undefined) return; + yield* orchestrator.dispatch({ + type: "checkpoint.rollback", + commandId: CommandId.make("rollback"), + threadId, + scopeId: threadStart.scopeId, + checkpointId: threadStart.id, + restoreFiles: false, + }); + yield* worker.drain(); + const rolledBack = yield* orchestrator.getThreadProjection(threadId); + assert.equal(rolledBack.runs[0]?.status, "rolled_back"); + assert.equal( + rolledBack.runs[0]?.delegatedCompletion?.delivery?.messageId, + steerMessageId, + ); + + yield* Queue.clear(workerReady); + yield* continuationGate.open; + yield* Queue.take(workerReady); + yield* worker.drain(); + const final = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + final.runs.map((run) => [run.id, run.status]), + [[parent.runId, "rolled_back"]], + ); + assert.equal(started.length, 1); + return; + } + if (!resultRead) { + const wakeA = yield* nextWake(1, { taskIds: [taskA] }); + assert.equal(wakeA.message.messageId, steerMessageId); + yield* settle(wakeA); + } + if (queuedMessage) { + const queued = yield* nextWake(1, { taskIds: [] }); + assert.equal(queued.message.messageId, queuedMessageId); + yield* settle(queued); + } + const wakeB = yield* nextWake(resultRead ? (queuedMessage ? 2 : 1) : 2, { + taskIds: [taskB], + }); + yield* settle(wakeB); + + const final = yield* orchestrator.getThreadProjection(threadId); + assert.equal(final.runs[0]?.delegatedCompletion?.delivery, null); + assert.deepEqual( + final.subagents.map((row) => [row.id, row.completionDelivery?.state]), + [ + [taskA, resultRead ? "acknowledged" : "delivered"], + [taskB, "delivered"], + ], + ); + assert.equal(steerCalls, 5); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { name }, + ProviderAdapterRegistry.layerSingle(adapter), + { runEffectWorker: false, runContinuationWorker: true, continuationRequests }, + ), + ), + ); + }), + ); +}); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index e3f5f72665b3..6bd23892d42a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4587,6 +4587,34 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } } + // A rollback discarded the run these results belong to, or the run a + // steer carried them into, so they no longer wake the thread. Accept the + // delivery without changing anything, before the wake would unsettle the + // thread, so the continuation worker does not retry it. + if (command.delegatedCompletion !== undefined) { + const { parentRunId } = command.delegatedCompletion; + const steeredRunId = projection.messages.find( + (message) => message.id === command.messageId, + )?.runId; + if ( + projection.runs.some( + (run) => + run.status === "rolled_back" && (run.id === parentRunId || run.id === steeredRunId), + ) + ) { + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + occurredAt: yield* DateTime.now, + payload: projection.thread, + }); + return; + } + } + const continuedTaskId = projection.runs.find( (run) => run.id === command.restartContinuationOfRunId, )?.delegatedTaskId; @@ -10437,6 +10465,163 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + /** + * Replaces a cohort's delivery with one reserving its pending siblings, or + * releases it when none are pending. `accepted` names the results a provider + * just took, which become delivered. Returns the events to write, in order, + * and the new reservation. + */ + const reserveNextDelegatedDelivery = (input: { + readonly projection: Pick; + readonly parentRun: OrchestrationV2Run; + readonly cohort: OrchestrationV2DelegatedCompletionCohort; + readonly accepted?: ReadonlyArray; + /** While the parent is live, settled_only results keep waiting. */ + readonly parentIsLive: boolean; + readonly now: DateTime.Utc; + }) => + Effect.gen(function* () { + const { projection, parentRun, cohort, parentIsLive, now } = input; + const threadId = projection.thread.id; + const pendingTaskIds = + projection.thread.archivedAt === null && projection.thread.deletedAt === null + ? projection.subagents + .filter( + (task) => + task.origin === "app_owned" && + task.runId === parentRun.id && + task.completionDelivery?.state === "pending" && + isTerminalDelegatedTaskStatus(task.status) && + (!parentIsLive || task.completionWake === "always"), + ) + .map((task) => task.id) + : []; + const nextDelivery = + pendingTaskIds.length === 0 + ? null + : { + generation: cohort.nextGeneration, + messageId: yield* mapDelegatedCompletionError( + idAllocator.allocate.message({ + threadId, + ordinal: + (yield* mapDelegatedCompletionError( + projectionStore.getMessageCount(threadId), + )) + 1, + }), + ), + taskIds: pendingTaskIds, + }; + const acceptedIds = new Set(input.accepted ?? []); + const pendingIds = new Set(pendingTaskIds); + const events: Array> = []; + for (const task of projection.subagents) { + const state = pendingIds.has(task.id) + ? "claimed" + : acceptedIds.has(task.id) && task.completionDelivery?.state === "claimed" + ? "delivered" + : undefined; + if (state === undefined) continue; + events.push({ + type: "subagent.updated", + threadId, + runId: parentRun.id, + nodeId: task.id, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + occurredAt: now, + payload: { + ...task, + completionDelivery: { state, observedByRunId: null }, + updatedAt: now, + }, + }); + } + events.push({ + type: "run.updated", + threadId, + runId: parentRun.id, + providerInstanceId: parentRun.providerInstanceId, + occurredAt: now, + payload: { + ...parentRun, + delegatedCompletion: { + ...cohort, + nextGeneration: cohort.nextGeneration + (nextDelivery === null ? 0 : 1), + delivery: nextDelivery, + }, + }, + }); + return { events, nextDelivery }; + }); + + /** + * A provider can reject every steer retry before its terminal event reaches + * the projection, leaving a delegated delivery reserved as a steer on a run + * that has since settled. Nothing else offers it again, and it blocks later + * siblings. Offer it as a wake; when the run already read every result in + * it, release it and reserve the pending siblings instead. A rolled-back + * run's deliveries stay discarded. + * + * Safe to repeat for one settlement, as a later terminal `run.updated` for the + * same run does. A release replaces the delivery, so the message and + * generation guard skips it next time. A repeated offer is dropped by the + * continuation worker and by message.dispatch once the wake has taken the + * message, and that wake also moves the message off this run. + */ + const recoverSettledDelegatedSteers = (threadId: ThreadId, settledRun: OrchestrationV2Run) => + Effect.gen(function* () { + if (settledRun.status === "rolled_back") return; + const { messages } = yield* projectionStore.getThreadRecords(threadId, ["messages"], { + messageRunIds: [settledRun.id], + messageRoles: ["user"], + }); + const steers = messages.filter( + (message) => + message.delegatedCompletion !== undefined && message.id !== settledRun.userMessageId, + ); + if (steers.length === 0) return; + const projection = yield* projectionStore.getThreadRecords(threadId, ["runs", "subagents"]); + // A rollback can commit between this terminal event and the thread lock. + if (projection.runs.find((run) => run.id === settledRun.id)?.status === "rolled_back") { + return; + } + for (const steer of steers) { + const ownership = steer.delegatedCompletion!; + const parentRun = projection.runs.find((run) => run.id === ownership.parentRunId); + const cohort = parentRun?.delegatedCompletion; + const delivery = cohort?.delivery; + if ( + parentRun === undefined || + parentRun.status === "rolled_back" || + cohort?.disposition !== "open" || + delivery == null || + delivery.messageId !== steer.id || + delivery.generation !== ownership.generation + ) { + continue; + } + if (delivery.taskIds.length > 0) { + yield* offerDelegatedCompletionDelivery(threadId, parentRun.id); + continue; + } + // The receiving run settled, so results pending behind this delivery are + // due a wake now, as in finalizeDelegatedCompletionDelivery. A message + // promoted from the queue since then must not hold settled_only ones. + const { events, nextDelivery } = yield* reserveNextDelegatedDelivery({ + projection, + parentRun, + cohort, + parentIsLive: false, + now: yield* DateTime.now, + }); + yield* writeSystemEvents(events); + if (nextDelivery !== null) { + yield* offerDelegatedCompletionDelivery(threadId, parentRun.id); + } + } + }); + const dispatchNotificationAccepted = Effect.fn("orchestrationV2.notificationAccepted")(function* ( command: Extract, events: Ref.Ref>, @@ -10465,74 +10650,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); return; } - // settled_only tasks defer only to a blocking wait in their spawning run. - const parentIsLive = hasLiveRun({ runs: [parentRun] }); - const pendingTaskIds = - projection.thread.archivedAt === null && projection.thread.deletedAt === null - ? projection.subagents - .filter( - (task) => - task.origin === "app_owned" && - task.runId === parentRun.id && - task.completionDelivery?.state === "pending" && - isTerminalDelegatedTaskStatus(task.status) && - (!parentIsLive || task.completionWake === "always"), - ) - .map((task) => task.id) - : []; - const nextDelivery = - pendingTaskIds.length === 0 - ? null - : { - generation: cohort.nextGeneration, - messageId: yield* mapDelegatedCompletionError( - idAllocator.allocate.message({ - threadId: command.threadId, - ordinal: - (yield* mapDelegatedCompletionError( - projectionStore.getMessageCount(projection.thread.id), - )) + 1, - }), - ), - taskIds: pendingTaskIds, - }; - const acceptedIds = new Set(delivery.taskIds); - const pendingIds = new Set(pendingTaskIds); - for (const task of projection.subagents) { - const state = pendingIds.has(task.id) - ? "claimed" - : acceptedIds.has(task.id) && task.completionDelivery?.state === "claimed" - ? "delivered" - : undefined; - if (state === undefined) continue; - yield* emitEvent({ - type: "subagent.updated", - threadId: command.threadId, - runId: parentRun.id, - nodeId: task.id, - driver: task.driver, - providerInstanceId: task.providerInstanceId, - occurredAt: now, - payload: { ...task, completionDelivery: { state, observedByRunId: null }, updatedAt: now }, - }); - } // Provider acceptance drains this batch but does not acknowledge its results. // task_status owns acknowledgment. - yield* emitEvent({ - type: "run.updated", - threadId: command.threadId, - runId: parentRun.id, - providerInstanceId: parentRun.providerInstanceId, - occurredAt: now, - payload: { - ...parentRun, - delegatedCompletion: { - ...cohort, - nextGeneration: cohort.nextGeneration + (nextDelivery === null ? 0 : 1), - delivery: nextDelivery, - }, - }, + const { events: reservationEvents } = yield* reserveNextDelegatedDelivery({ + projection, + parentRun, + cohort, + accepted: delivery.taskIds, + // settled_only tasks defer only to a blocking wait in their spawning run. + parentIsLive: hasLiveRun({ runs: [parentRun] }), + now, }); + for (const event of reservationEvents) { + yield* emitEvent(event); + } }); const dispatchUnsupported = (command: OrchestrationV2ServerCommand) => @@ -11115,6 +11246,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if (parentThreadId !== undefined) { yield* threadDispatch.withLock(parentThreadId, finalizeAppOwnedSubagent(threadId)); } + if (stored.event.type === "run.updated") { + yield* threadDispatch.withLock( + threadId, + recoverSettledDelegatedSteers(threadId, stored.event.payload), + ); + } }).pipe( Effect.catchCause((cause) => Effect.logWarning("Failed to react to terminal V2 run", { diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 27c1cc28ae5a..09f1434701c9 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -268,6 +268,9 @@ export function layerWithRegistry( // Start continuation runs for provider wake turns, as the live runtime does. // Off by default: most fixtures record no wake turn. readonly runContinuationWorker?: boolean; + // Replaces the continuation request queue, for example to hold the + // continuation worker between an offer and its dispatch. + readonly continuationRequests?: typeof ProviderContinuationRequests.layer; // Reconcile a previous runtime's state before the effect worker starts, // as server startup does after a crash or restart. readonly recoverOnStartup?: boolean; @@ -294,7 +297,9 @@ export function layerWithRegistry( // One queue shared by the adapters, the orchestrator, and the worker, like // runtimeLayer.ts; layer memoization keeps it a single instance. const layerContinuationRequests = - options.runContinuationWorker === true ? ProviderContinuationRequests.layer : Layer.empty; + options.runContinuationWorker === true + ? (options.continuationRequests ?? ProviderContinuationRequests.layer) + : Layer.empty; const layerProvidedRegistry = registryLayer.pipe( Layer.provide(Layer.merge(layerContinuationRequests, McpProviderSessions.layer)), );