From 41e0542636581e9c9502a965ba5557cb9d4eea6d Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:11:14 -0700 Subject: [PATCH 1/7] fix(server): a delegated result whose steer never landed wakes the parent after it settles Co-Authored-By: Claude Opus 5.5 (1M context) --- ...legatedSteerSettlement.integration.test.ts | 391 ++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 113 +++++ 2 files changed, 504 insertions(+) create mode 100644 apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts 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..d3ebaffec621 --- /dev/null +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -0,0 +1,391 @@ +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + ProviderTurnId, + type RunId, + ThreadId, + type OrchestrationV2DomainEvent, + type OrchestrationV2Subagent, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +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 { + ProviderAdapterSteerRunError, + type ProviderAdapterV2Event, + type ProviderAdapterV2Shape, + type ProviderAdapterV2TurnInput, +} from "./ProviderAdapter.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; + +const driver = ProviderDriverKind.make("codex"); +const instanceId = ProviderInstanceId.make("codex"); +const modelSelection = { instanceId, model: "test-model" }; +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. +for (const resultRead of [false, true]) { + it.effect( + resultRead + ? "releases a steered delegated delivery the parent already read when it settles" + : "re-offers a steered delegated delivery whose steer never landed when the parent settles", + () => + Effect.scoped( + Effect.gen(function* () { + const name = `delegated-steer-settlement-${resultRead ? "read" : "unaccepted"}`; + const cwd = yield* checkpointWorkspace(name); + const events = yield* Queue.unbounded(); + const started: ProviderAdapterV2TurnInput[] = []; + let steerCalls = 0; + const adapter: ProviderAdapterV2Shape = { + 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 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: () => Effect.die("unused"), + 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}`); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + // Ends the run's provider turn and waits for the run to settle. + const settle = (runId: RunId, runOrdinal: number) => + Effect.gen(function* () { + const waiting = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === "waiting", + ); + const completed = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === "completed", + ); + const projection = yield* orchestrator.getThreadProjection(threadId); + const run = projection.runs.find((candidate) => candidate.id === runId)!; + const turn = projection.providerTurns.find( + (candidate) => candidate.runAttemptId === run.activeAttemptId, + )!; + 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, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(waiting); + yield* worker.drain(); + yield* Fiber.join(completed); + }); + // Subscribes, before the settlement that triggers it, to the wake run + // the continuation worker starts; joining it runs that wake's turn. + const watchWake = (startedCount: number) => + Effect.gen(function* () { + const wake = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.ordinal === startedCount + 1 && + event.payload.status !== "queued", + ); + return Effect.gen(function* () { + yield* Fiber.join(wake); + yield* worker.drain(); + assert.equal(started.length, startedCount + 1); + return started[startedCount]!; + }); + }); + + 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", + }); + const running = yield* watch( + (event) => + event.type === "provider-turn.updated" && event.payload.status === "running", + ); + 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(); + yield* Fiber.join(running); + const parent = started[0]!; + + const steerMessageId = MessageId.make("message:delegated-steer"); + const taskA = NodeId.make("task:a"); + const taskB = NodeId.make("task:b"); + 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"], + ): 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: "always", + 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 }), + task(taskB, { state: "pending", observedByRunId: null }), + ].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. + while (steerCalls < 5) { + 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 runWakeB = (wakeB: ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + assert.include(wakeB.message.text, String(taskB)); + assert.notInclude(wakeB.message.text, String(taskA)); + yield* settle(wakeB.runId, wakeB.runOrdinal); + }); + const beforeSettle = yield* orchestrator.getThreadProjection(threadId); + assert.equal(beforeSettle.runs[0]?.status, "running"); + assert.equal( + beforeSettle.runs[0]?.delegatedCompletion?.delivery?.messageId, + steerMessageId, + ); + + const firstWake = yield* watchWake(1); + yield* settle(parent.runId, parent.runOrdinal); + const wakeA = resultRead ? undefined : yield* firstWake; + if (wakeA !== undefined) { + assert.equal(wakeA.message.messageId, steerMessageId); + assert.include(wakeA.message.text, String(taskA)); + assert.notInclude(wakeA.message.text, String(taskB)); + const secondWake = yield* watchWake(2); + yield* settle(wakeA.runId, wakeA.runOrdinal); + yield* runWakeB(yield* secondWake); + } else { + yield* runWakeB(yield* firstWake); + } + + 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( + makeOrchestratorV2ReplayLayerWithRegistry( + { name }, + ProviderAdapterRegistry.makeSingleLayer(adapter), + { runEffectWorker: false, runContinuationWorker: true }, + ), + ), + ); + }), + ), + ); +} diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index de100d291b8f..ec163f27ca25 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9938,6 +9938,113 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + /** + * 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. + */ + 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"]); + 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; + } + const parentIsLive = hasLiveRun(projection); + const pendingTasks = + 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"), + ) + : []; + const nextDelivery = + pendingTasks.length === 0 + ? null + : { + generation: cohort.nextGeneration, + messageId: yield* mapDelegatedCompletionError( + idAllocator.allocate.message({ + threadId, + ordinal: + (yield* mapDelegatedCompletionError( + projectionStore.getMessageCount(threadId), + )) + 1, + }), + ), + taskIds: pendingTasks.map((task) => task.id), + }; + const now = yield* DateTime.now; + yield* writeSystemEvents([ + ...pendingTasks.map((task) => ({ + type: "subagent.updated" as const, + threadId, + runId: parentRun.id, + 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", + threadId, + runId: parentRun.id, + ...(parentRun.rootNodeId === null ? {} : { nodeId: parentRun.rootNodeId }), + providerInstanceId: parentRun.providerInstanceId, + occurredAt: now, + payload: { + ...parentRun, + delegatedCompletion: { + ...cohort, + nextGeneration: cohort.nextGeneration + (nextDelivery === null ? 0 : 1), + delivery: nextDelivery, + }, + }, + }, + ]); + if (nextDelivery !== null) { + yield* offerDelegatedCompletionDelivery(threadId, parentRun.id); + } + } + }); + const dispatchNotificationAccepted = Effect.fn("orchestrationV2.notificationAccepted")(function* ( command: Extract, events: Ref.Ref>, @@ -10563,6 +10670,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : undefined, ), ); + 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", { From a84b542950314c0ccfe58484402830151899be29 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:19:14 -0700 Subject: [PATCH 2/7] test(server): assert the stranded delegated delivery instead of timing out Co-Authored-By: Claude Opus 5.5 (1M context) --- ...legatedSteerSettlement.integration.test.ts | 76 +++++++++++-------- 1 file changed, 43 insertions(+), 33 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts index d3ebaffec621..b209eba0094b 100644 --- a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -12,6 +12,7 @@ import { type RunId, ThreadId, type OrchestrationV2DomainEvent, + type OrchestrationV2ProviderTurn, type OrchestrationV2Subagent, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; @@ -37,6 +38,9 @@ import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; const driver = ProviderDriverKind.make("codex"); const instanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId, model: "test-model" }; +const yieldToRuntime = Effect.yieldNow.pipe( + Effect.andThen(Effect.promise(() => new Promise((resolve) => setImmediate(resolve)))), +); const capabilities = { ...CodexProviderCapabilitiesV2, turns: { ...CodexProviderCapabilitiesV2.turns, supportsActiveSteering: true }, @@ -172,11 +176,20 @@ for (const resultRead of [false, true]) { event.payload.id === runId && event.payload.status === "completed", ); - const projection = yield* orchestrator.getThreadProjection(threadId); - const run = projection.runs.find((candidate) => candidate.id === runId)!; - const turn = projection.providerTurns.find( - (candidate) => candidate.runAttemptId === run.activeAttemptId, - )!; + // startTurn's running turn is ingested on the adapter event fiber. + let turn: OrchestrationV2ProviderTurn | undefined; + for (let attempt = 0; attempt < 200 && turn === undefined; attempt++) { + const projection = yield* orchestrator.getThreadProjection(threadId); + const run = projection.runs.find((candidate) => candidate.id === runId); + turn = projection.providerTurns.find( + (candidate) => + candidate.runAttemptId === run?.activeAttemptId && + candidate.status === "running", + ); + if (turn === undefined) yield* yieldToRuntime; + } + assert.isDefined(turn, `run ${runId} has no running provider turn`); + if (turn === undefined) return; yield* Queue.offer(events, { type: "provider_turn.updated", driver, @@ -196,22 +209,26 @@ for (const resultRead of [false, true]) { yield* worker.drain(); yield* Fiber.join(completed); }); - // Subscribes, before the settlement that triggers it, to the wake run - // the continuation worker starts; joining it runs that wake's turn. - const watchWake = (startedCount: number) => + // The continuation worker runs on its own fiber, so drain the effect + // worker and yield to it until nothing new starts. A wake that never + // starts then fails the assertion with the stranded cohort state. + const nextWake = (startedCount: number) => Effect.gen(function* () { - const wake = yield* watch( - (event) => - event.type === "run.updated" && - event.payload.ordinal === startedCount + 1 && - event.payload.status !== "queued", - ); - return Effect.gen(function* () { - yield* Fiber.join(wake); + for (let attempt = 0; attempt < 200 && started.length === startedCount; attempt++) { yield* worker.drain(); - assert.equal(started.length, startedCount + 1); - return started[startedCount]!; - }); + if (started.length === startedCount) yield* yieldToRuntime; + } + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + started.length, + startedCount + 1, + `no wake started: delivery=${JSON.stringify( + projection.runs[0]?.delegatedCompletion?.delivery, + )} tasks=${JSON.stringify( + projection.subagents.map((row) => [row.id, row.completionDelivery?.state]), + )}`, + ); + return started[startedCount]!; }); yield* orchestrator.dispatch({ @@ -323,7 +340,7 @@ for (const resultRead of [false, true]) { delegatedCompletion: { parentRunId: parent.runId, generation: 1, taskIds: [taskA] }, }); // Every retry is rejected while the parent turn still looks running. - while (steerCalls < 5) { + for (let attempt = 0; attempt < 5; attempt++) { yield* worker.drain(); yield* TestClock.adjust("1 minute"); } @@ -339,12 +356,6 @@ for (const resultRead of [false, true]) { observedByRunId: parent.runId, }); } - const runWakeB = (wakeB: ProviderAdapterV2TurnInput) => - Effect.gen(function* () { - assert.include(wakeB.message.text, String(taskB)); - assert.notInclude(wakeB.message.text, String(taskA)); - yield* settle(wakeB.runId, wakeB.runOrdinal); - }); const beforeSettle = yield* orchestrator.getThreadProjection(threadId); assert.equal(beforeSettle.runs[0]?.status, "running"); assert.equal( @@ -352,19 +363,18 @@ for (const resultRead of [false, true]) { steerMessageId, ); - const firstWake = yield* watchWake(1); yield* settle(parent.runId, parent.runOrdinal); - const wakeA = resultRead ? undefined : yield* firstWake; - if (wakeA !== undefined) { + if (!resultRead) { + const wakeA = yield* nextWake(1); assert.equal(wakeA.message.messageId, steerMessageId); assert.include(wakeA.message.text, String(taskA)); assert.notInclude(wakeA.message.text, String(taskB)); - const secondWake = yield* watchWake(2); yield* settle(wakeA.runId, wakeA.runOrdinal); - yield* runWakeB(yield* secondWake); - } else { - yield* runWakeB(yield* firstWake); } + const wakeB = yield* nextWake(resultRead ? 1 : 2); + assert.include(wakeB.message.text, String(taskB)); + assert.notInclude(wakeB.message.text, String(taskA)); + yield* settle(wakeB.runId, wakeB.runOrdinal); const final = yield* orchestrator.getThreadProjection(threadId); assert.equal(final.runs[0]?.delegatedCompletion?.delivery, null); From 015cbdbb6f41c84bbecc055fd18c9199ed8559f7 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:23:18 -0700 Subject: [PATCH 3/7] test(server): show which delegated delivery a missing wake left reserved Co-Authored-By: Claude Opus 5.5 (1M context) --- ...legatedSteerSettlement.integration.test.ts | 51 +++++++++++-------- 1 file changed, 29 insertions(+), 22 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts index b209eba0094b..d412165b0567 100644 --- a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -38,6 +38,8 @@ import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.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 yieldToRuntime = Effect.yieldNow.pipe( Effect.andThen(Effect.promise(() => new Promise((resolve) => setImmediate(resolve)))), ); @@ -210,25 +212,36 @@ for (const resultRead of [false, true]) { yield* Fiber.join(completed); }); // The continuation worker runs on its own fiber, so drain the effect - // worker and yield to it until nothing new starts. A wake that never - // starts then fails the assertion with the stranded cohort state. - const nextWake = (startedCount: number) => + // worker and yield to it until a wake starts. When none does, the + // assertions show the delivery reservation left behind. + const nextWake = ( + startedCount: number, + expected: { readonly taskIds: ReadonlyArray }, + ) => Effect.gen(function* () { for (let attempt = 0; attempt < 200 && started.length === startedCount; attempt++) { yield* worker.drain(); if (started.length === startedCount) yield* yieldToRuntime; } - const projection = yield* orchestrator.getThreadProjection(threadId); - assert.equal( - started.length, - startedCount + 1, - `no wake started: delivery=${JSON.stringify( - projection.runs[0]?.delegatedCompletion?.delivery, - )} tasks=${JSON.stringify( - projection.subagents.map((row) => [row.id, row.completionDelivery?.state]), - )}`, - ); - return started[startedCount]!; + if (started.length === startedCount) { + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.runs[0]?.delegatedCompletion?.delivery?.taskIds, + expected.taskIds, + "no wake started for the reserved delivery", + ); + assert.fail("no wake started for the reserved delivery"); + } + const wake = started[startedCount]!; + assert.equal(started.length, startedCount + 1); + 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({ @@ -265,8 +278,6 @@ for (const resultRead of [false, true]) { const parent = started[0]!; const steerMessageId = MessageId.make("message:delegated-steer"); - const taskA = NodeId.make("task:a"); - const taskB = NodeId.make("task:b"); const projection = yield* orchestrator.getThreadProjection(threadId); const parentRun = projection.runs.find((run) => run.id === parent.runId)!; const now = yield* DateTime.now; @@ -365,15 +376,11 @@ for (const resultRead of [false, true]) { yield* settle(parent.runId, parent.runOrdinal); if (!resultRead) { - const wakeA = yield* nextWake(1); + const wakeA = yield* nextWake(1, { taskIds: [taskA] }); assert.equal(wakeA.message.messageId, steerMessageId); - assert.include(wakeA.message.text, String(taskA)); - assert.notInclude(wakeA.message.text, String(taskB)); yield* settle(wakeA.runId, wakeA.runOrdinal); } - const wakeB = yield* nextWake(resultRead ? 1 : 2); - assert.include(wakeB.message.text, String(taskB)); - assert.notInclude(wakeB.message.text, String(taskA)); + const wakeB = yield* nextWake(resultRead ? 1 : 2, { taskIds: [taskB] }); yield* settle(wakeB.runId, wakeB.runOrdinal); const final = yield* orchestrator.getThreadProjection(threadId); From 3405ef85b8b0406f49d22627f8c71cf2103c3906 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:30:16 -0700 Subject: [PATCH 4/7] refactor(server): share the next delegated reservation between acceptance and settled steers Co-Authored-By: Claude Opus 5.5 (1M context) --- .../src/orchestration-v2/Orchestrator.ts | 238 +++++++++--------- 1 file changed, 112 insertions(+), 126 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ec163f27ca25..eaa9123422a6 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9938,6 +9938,96 @@ 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; + readonly now: DateTime.Utc; + }) => + Effect.gen(function* () { + const { projection, parentRun, cohort, now } = input; + const threadId = projection.thread.id; + // 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, + 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 @@ -9945,6 +10035,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio * 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* () { @@ -9978,67 +10074,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* offerDelegatedCompletionDelivery(threadId, parentRun.id); continue; } - const parentIsLive = hasLiveRun(projection); - const pendingTasks = - 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"), - ) - : []; - const nextDelivery = - pendingTasks.length === 0 - ? null - : { - generation: cohort.nextGeneration, - messageId: yield* mapDelegatedCompletionError( - idAllocator.allocate.message({ - threadId, - ordinal: - (yield* mapDelegatedCompletionError( - projectionStore.getMessageCount(threadId), - )) + 1, - }), - ), - taskIds: pendingTasks.map((task) => task.id), - }; - const now = yield* DateTime.now; - yield* writeSystemEvents([ - ...pendingTasks.map((task) => ({ - type: "subagent.updated" as const, - threadId, - runId: parentRun.id, - 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", - threadId, - runId: parentRun.id, - ...(parentRun.rootNodeId === null ? {} : { nodeId: parentRun.rootNodeId }), - providerInstanceId: parentRun.providerInstanceId, - occurredAt: now, - payload: { - ...parentRun, - delegatedCompletion: { - ...cohort, - nextGeneration: cohort.nextGeneration + (nextDelivery === null ? 0 : 1), - delivery: nextDelivery, - }, - }, - }, - ]); + const { events, nextDelivery } = yield* reserveNextDelegatedDelivery({ + projection, + parentRun, + cohort, + now: yield* DateTime.now, + }); + yield* writeSystemEvents(events); if (nextDelivery !== null) { yield* offerDelegatedCompletionDelivery(threadId, parentRun.id); } @@ -10073,74 +10115,18 @@ 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, + now, }); + for (const event of reservationEvents) { + yield* emitEvent(event); + } }); const dispatchUnsupported = (command: OrchestrationV2ServerCommand) => From 4f7304eb038afc9fcf4320164b43671cba8e2ebd Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Thu, 1 Oct 2026 23:36:00 -0700 Subject: [PATCH 5/7] fix(server): settled delegated steers reserve settled-only siblings and skip rolled-back runs Co-Authored-By: Claude Opus 5.5 (1M context) --- ...legatedSteerSettlement.integration.test.ts | 715 +++++++++--------- .../src/orchestration-v2/Orchestrator.ts | 16 +- 2 files changed, 391 insertions(+), 340 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts index d412165b0567..0a1209264fda 100644 --- a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -32,7 +32,7 @@ import { type ProviderAdapterV2TurnInput, } from "./ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; -import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; +import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; const driver = ProviderDriverKind.make("codex"); @@ -53,356 +53,397 @@ const capabilities = { // 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. -for (const resultRead of [false, true]) { - it.effect( - resultRead - ? "releases a steered delegated delivery the parent already read when it settles" - : "re-offers a steered delegated delivery whose steer never landed when the parent settles", - () => - Effect.scoped( - Effect.gen(function* () { - const name = `delegated-steer-settlement-${resultRead ? "read" : "unaccepted"}`; - const cwd = yield* checkpointWorkspace(name); - const events = yield* Queue.unbounded(); - const started: ProviderAdapterV2TurnInput[] = []; - let steerCalls = 0; - const adapter: ProviderAdapterV2Shape = { - 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, +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, + }, +] as const)("$title", (scenario) => { + const { resultRead, queuedMessage } = scenario; + 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: ProviderAdapterV2TurnInput[] = []; + let steerCalls = 0; + const adapter: ProviderAdapterV2Shape = { + 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, - providerSession: { - id: 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, - 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 ProviderAdapterSteerRunError({ + providerTurn: { + id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), + providerThreadId: turn.providerThread.id, + nodeId: turn.rootNodeId, + runAttemptId: turn.attemptId, + nativeTurnRef: { driver, - providerThreadId: turn.providerThread.id, - providerTurnId: turn.providerTurnId, - cause: "turn is not active", - }); - }), - interruptTurn: () => Effect.void, - respondToRuntimeRequest: () => Effect.void, - readThreadSnapshot: () => Effect.die("unused"), - rollbackThread: () => Effect.die("unused"), - 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}`); - const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => - orchestrator.streamDomainEvents.pipe( - Stream.filter(predicate), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - // Ends the run's provider turn and waits for the run to settle. - const settle = (runId: RunId, runOrdinal: number) => - Effect.gen(function* () { - const waiting = yield* watch( - (event) => - event.type === "run.updated" && - event.payload.id === runId && - event.payload.status === "waiting", - ); - const completed = yield* watch( - (event) => - event.type === "run.updated" && - event.payload.id === runId && - event.payload.status === "completed", - ); - // startTurn's running turn is ingested on the adapter event fiber. - let turn: OrchestrationV2ProviderTurn | undefined; - for (let attempt = 0; attempt < 200 && turn === undefined; attempt++) { - const projection = yield* orchestrator.getThreadProjection(threadId); - const run = projection.runs.find((candidate) => candidate.id === runId); - turn = projection.providerTurns.find( - (candidate) => - candidate.runAttemptId === run?.activeAttemptId && - candidate.status === "running", - ); - if (turn === undefined) yield* yieldToRuntime; - } - assert.isDefined(turn, `run ${runId} has no running provider turn`); - if (turn === undefined) return; - 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, - status: "completed", - failure: null, - threadDisposition: "reusable", - }); - yield* Fiber.join(waiting); - yield* worker.drain(); - yield* Fiber.join(completed); - }); - // The continuation worker runs on its own fiber, so drain the effect - // worker and yield to it until a wake starts. When none does, the - // assertions show the delivery reservation left behind. - const nextWake = ( - startedCount: number, - expected: { readonly taskIds: ReadonlyArray }, - ) => - Effect.gen(function* () { - for (let attempt = 0; attempt < 200 && started.length === startedCount; attempt++) { - yield* worker.drain(); - if (started.length === startedCount) yield* yieldToRuntime; - } - if (started.length === startedCount) { - const projection = yield* orchestrator.getThreadProjection(threadId); - assert.deepEqual( - projection.runs[0]?.delegatedCompletion?.delivery?.taskIds, - expected.taskIds, - "no wake started for the reserved delivery", - ); - assert.fail("no wake started for the reserved delivery"); - } - const wake = started[startedCount]!; - assert.equal(started.length, startedCount + 1); - 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", - }); - const running = yield* watch( + 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 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: () => Effect.die("unused"), + 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}`); + const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => + orchestrator.streamDomainEvents.pipe( + Stream.filter(predicate), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + // Ends the run's provider turn and waits for the run to settle. + const settle = (runId: RunId, runOrdinal: number) => + Effect.gen(function* () { + const waiting = yield* watch( (event) => - event.type === "provider-turn.updated" && event.payload.status === "running", + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === "waiting", ); - 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", + const completed = yield* watch( + (event) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === "completed", + ); + // startTurn's running turn is ingested on the adapter event fiber. + let turn: OrchestrationV2ProviderTurn | undefined; + for (let attempt = 0; attempt < 200 && turn === undefined; attempt++) { + const projection = yield* orchestrator.getThreadProjection(threadId); + const run = projection.runs.find((candidate) => candidate.id === runId); + turn = projection.providerTurns.find( + (candidate) => + candidate.runAttemptId === run?.activeAttemptId && candidate.status === "running", + ); + if (turn === undefined) yield* yieldToRuntime; + } + assert.isDefined(turn, `run ${runId} has no running provider turn`); + if (turn === undefined) return; + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, }); - yield* worker.drain(); - yield* Fiber.join(running); - const parent = started[0]!; - - 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"], - ): OrchestrationV2Subagent => ({ - id, - threadId, - runId: parent.runId, - parentNodeId: parent.rootNodeId, - origin: "app_owned", - createdBy: "agent", + yield* Queue.offer(events, { + type: "turn.terminal", driver, - providerInstanceId: instanceId, - providerThreadId: null, - childThreadId: null, - nativeTaskRef: null, - prompt: `Do ${id}`, - title: null, - model: null, - completionWake: "always", - completionDelivery, + providerThreadId: turn.providerThreadId, + providerTurnId: turn.id, + runOrdinal, 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 }), - task(taskB, { state: "pending", observedByRunId: null }), - ].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] }, + failure: null, + threadDisposition: "reusable", }); - // Every retry is rejected while the parent turn still looks running. - for (let attempt = 0; attempt < 5; attempt++) { + yield* Fiber.join(waiting); + yield* worker.drain(); + yield* Fiber.join(completed); + }); + // The continuation worker runs on its own fiber, so drain the effect + // worker and yield to it until a wake starts. When none does, the + // assertions show the delivery reservation left behind. + const nextWake = ( + startedCount: number, + expected: { readonly taskIds: ReadonlyArray }, + ) => + Effect.gen(function* () { + for (let attempt = 0; attempt < 200 && started.length === startedCount; attempt++) { yield* worker.drain(); - yield* TestClock.adjust("1 minute"); + if (started.length === startedCount) yield* yieldToRuntime; } - 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, - }); + if (started.length === startedCount) { + const projection = yield* orchestrator.getThreadProjection(threadId); + assert.deepEqual( + projection.runs[0]?.delegatedCompletion?.delivery?.taskIds, + expected.taskIds, + "no wake started for the reserved delivery", + ); + assert.fail("no wake started for the reserved delivery"); } - 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.runId, parent.runOrdinal); - if (!resultRead) { - const wakeA = yield* nextWake(1, { taskIds: [taskA] }); - assert.equal(wakeA.message.messageId, steerMessageId); - yield* settle(wakeA.runId, wakeA.runOrdinal); + const wake = started[startedCount]!; + assert.equal(started.length, startedCount + 1); + 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)); + } } - const wakeB = yield* nextWake(resultRead ? 1 : 2, { taskIds: [taskB] }); - yield* settle(wakeB.runId, wakeB.runOrdinal); + return wake; + }); - 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( - makeOrchestratorV2ReplayLayerWithRegistry( - { name }, - ProviderAdapterRegistry.makeSingleLayer(adapter), - { runEffectWorker: false, runContinuationWorker: true }, + 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", + }); + const running = yield* watch( + (event) => event.type === "provider-turn.updated" && event.payload.status === "running", + ); + 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(); + yield* Fiber.join(running); + const parent = started[0]!; + + 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.runId, parent.runOrdinal); + if (!resultRead) { + const wakeA = yield* nextWake(1, { taskIds: [taskA] }); + assert.equal(wakeA.message.messageId, steerMessageId); + yield* settle(wakeA.runId, wakeA.runOrdinal); + } + if (queuedMessage) { + const queued = yield* nextWake(1, { taskIds: [] }); + assert.equal(queued.message.messageId, queuedMessageId); + yield* settle(queued.runId, queued.runOrdinal); + } + const wakeB = yield* nextWake(resultRead ? (queuedMessage ? 2 : 1) : 2, { + taskIds: [taskB], + }); + yield* settle(wakeB.runId, wakeB.runOrdinal); + + 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 }, + ), + ), + ); + }), ); -} +}); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index eaa9123422a6..1dc7834a4e93 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -9949,13 +9949,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio 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, now } = input; + const { projection, parentRun, cohort, parentIsLive, now } = input; const threadId = projection.thread.id; - // 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 @@ -10055,6 +10055,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ); 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); @@ -10074,10 +10078,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio 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); @@ -10122,6 +10130,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio 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) { From ea90edb7e650d5a3890f8e987dd7f61d7eb18e46 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 6 Oct 2026 16:45:41 -0700 Subject: [PATCH 6/7] fix(server): a delegated wake whose turn was rolled back no longer starts Settlement recovery checks for a rollback before it offers a wake, but the continuation worker dispatches later. A rollback that lands in between left the parent rolled back while the released wake still started a turn. message.dispatch now drops a delegated completion whose parent run, or the run its steer was delivered into, was rolled back: it records an accepted receipt without starting a run or unsettling the thread, as for a stale restart continuation. The check sits in dispatch validation, so it covers every path that dispatches a delegated completion. The settlement test gains a case that holds the continuation worker after the recovered offer, rolls the parent back to the thread start, releases the worker, and checks that no second run starts. Co-Authored-By: Claude Opus 5.5 (1M context) --- ...legatedSteerSettlement.integration.test.ts | 86 ++++++++++++++++++- .../src/orchestration-v2/Orchestrator.ts | 28 ++++++ .../testkit/ProviderReplayHarness.ts | 7 +- 3 files changed, 118 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts index 0a1209264fda..a7ba2e31d9d1 100644 --- a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -18,6 +18,8 @@ import { import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as Latch from "effect/Latch"; +import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -32,6 +34,7 @@ import { type ProviderAdapterV2TurnInput, } from "./ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderContinuationRequests from "./ProviderContinuationRequests.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; @@ -75,8 +78,18 @@ it.effect.each([ 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}`; @@ -84,6 +97,31 @@ it.effect.each([ const events = yield* Queue.unbounded(); const started: 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: ProviderAdapterV2Shape = { instanceId, driver, @@ -164,7 +202,13 @@ it.effect.each([ interruptTurn: () => Effect.void, respondToRuntimeRequest: () => Effect.void, readThreadSnapshot: () => Effect.die("unused"), - rollbackThread: () => Effect.die("unused"), + rollbackThread: ({ providerThread }) => + Effect.succeed({ + providerThread, + providerTurns: [], + messages: [], + runtimeRequests: [], + }), forkThread: () => Effect.die("unused"), }; }), @@ -410,6 +454,44 @@ it.effect.each([ ); yield* settle(parent.runId, parent.runOrdinal); + if (rollback) { + // The settlement offered A's wake; the worker holds it at the gate. + const offer = yield* Queue.take(continuationOffers); + assert.equal(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); @@ -440,7 +522,7 @@ it.effect.each([ ProviderReplayHarness.layerWithRegistry( { name }, ProviderAdapterRegistry.layerSingle(adapter), - { runEffectWorker: false, runContinuationWorker: true }, + { runEffectWorker: false, runContinuationWorker: true, continuationRequests }, ), ), ); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 1dc7834a4e93..eb7cf01a4eb4 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4483,6 +4483,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; + } + } + if (projection.thread.settledOverride !== null) { const now = yield* DateTime.now; const thread: OrchestrationV2AppThread = { diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 687e881c4656..6f0388d70e3a 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -255,6 +255,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; @@ -281,7 +284,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(layerContinuationRequests)); const layerServerSettings = ServerSettings.layerTest({ responseStreamingMode: "turn", From f6864d88ed2ec38a896f5a586636ceefcf305618 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:03:21 -0700 Subject: [PATCH 7/7] test(server): settlement test waits on stored events instead of polling The delegated steer settlement test yielded to the runtime up to 200 times while it looked for a running provider turn and for the next wake. It now reads the thread's stored events through the event sink: the provider turn's running update, the run's waiting and completed updates, and the next run created or promoted as starting. A replayed stream also sees an event that landed before the wait began, such as a queued message promoted while the previous run settled. A live-clock bound turns a missing event into the same assertions as before, and the three cases still assert the same outcomes. Co-Authored-By: Claude Opus 5.5 (1M context) --- ...legatedSteerSettlement.integration.test.ts | 136 ++++++++++-------- 1 file changed, 77 insertions(+), 59 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts index a7ba2e31d9d1..070eb2d90411 100644 --- a/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedSteerSettlement.integration.test.ts @@ -9,17 +9,15 @@ import { ProviderInstanceId, ProviderThreadId, ProviderTurnId, - type RunId, ThreadId, type OrchestrationV2DomainEvent, - type OrchestrationV2ProviderTurn, type OrchestrationV2Subagent, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; -import * as Fiber from "effect/Fiber"; 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"; @@ -43,9 +41,6 @@ const instanceId = ProviderInstanceId.make("codex"); const modelSelection = { instanceId, model: "test-model" }; const taskA = NodeId.make("task:a"); const taskB = NodeId.make("task:b"); -const yieldToRuntime = Effect.yieldNow.pipe( - Effect.andThen(Effect.promise(() => new Promise((resolve) => setImmediate(resolve)))), -); const capabilities = { ...CodexProviderCapabilitiesV2, turns: { ...CodexProviderCapabilitiesV2.turns, supportsActiveSteering: true }, @@ -218,41 +213,44 @@ it.effect.each([ const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; const sink = yield* EventSink.EventSinkV2; const threadId = ThreadId.make(`thread:${name}`); - const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => - orchestrator.streamDomainEvents.pipe( - Stream.filter(predicate), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, + // 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), ); - // Ends the run's provider turn and waits for the run to settle. - const settle = (runId: RunId, runOrdinal: number) => + // startTurn's running turn is ingested on the adapter event fiber. + const runningTurn = (turn: ProviderAdapterV2TurnInput) => Effect.gen(function* () { - const waiting = yield* watch( + const running = yield* awaitEvent( + ["provider-turn.updated"], (event) => - event.type === "run.updated" && - event.payload.id === runId && - event.payload.status === "waiting", + event.payload.runAttemptId === turn.attemptId && event.payload.status === "running", ); - const completed = yield* watch( - (event) => - event.type === "run.updated" && - event.payload.id === runId && - event.payload.status === "completed", - ); - // startTurn's running turn is ingested on the adapter event fiber. - let turn: OrchestrationV2ProviderTurn | undefined; - for (let attempt = 0; attempt < 200 && turn === undefined; attempt++) { - const projection = yield* orchestrator.getThreadProjection(threadId); - const run = projection.runs.find((candidate) => candidate.id === runId); - turn = projection.providerTurns.find( - (candidate) => - candidate.runAttemptId === run?.activeAttemptId && candidate.status === "running", - ); - if (turn === undefined) yield* yieldToRuntime; + if (Option.isNone(running)) { + return assert.fail(`run ${turn.runId} has no running provider turn`); } - assert.isDefined(turn, `run ${runId} has no running provider turn`); - if (turn === undefined) return; + return running.value.payload; + }); + // Ends the run's provider turn and waits for the run to settle. + const settle = (input: ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + const turn = yield* runningTurn(input); yield* Queue.offer(events, { type: "provider_turn.updated", driver, @@ -263,38 +261,54 @@ it.effect.each([ driver, providerThreadId: turn.providerThreadId, providerTurnId: turn.id, - runOrdinal, + runOrdinal: input.runOrdinal, status: "completed", failure: null, threadDisposition: "reusable", }); - yield* Fiber.join(waiting); + 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* Fiber.join(completed); + yield* runReached("completed"); }); - // The continuation worker runs on its own fiber, so drain the effect - // worker and yield to it until a wake starts. When none does, the - // assertions show the delivery reservation left behind. + // 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* () { - for (let attempt = 0; attempt < 200 && started.length === startedCount; attempt++) { - yield* worker.drain(); - if (started.length === startedCount) yield* yieldToRuntime; - } - if (started.length === startedCount) { + 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", ); - assert.fail("no wake started for the reserved delivery"); + return assert.fail("no wake started for the reserved delivery"); } - const wake = started[startedCount]!; + 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)); @@ -319,9 +333,6 @@ it.effect.each([ createdBy: "user", creationSource: "web", }); - const running = yield* watch( - (event) => event.type === "provider-turn.updated" && event.payload.status === "running", - ); yield* orchestrator.dispatch({ type: "message.dispatch", commandId: CommandId.make("first"), @@ -334,8 +345,8 @@ it.effect.each([ creationSource: "web", }); yield* worker.drain(); - yield* Fiber.join(running); const parent = started[0]!; + yield* runningTurn(parent); const steerMessageId = MessageId.make("message:delegated-steer"); const projection = yield* orchestrator.getThreadProjection(threadId); @@ -453,11 +464,18 @@ it.effect.each([ steerMessageId, ); - yield* settle(parent.runId, parent.runOrdinal); + yield* settle(parent); if (rollback) { // The settlement offered A's wake; the worker holds it at the gate. - const offer = yield* Queue.take(continuationOffers); - assert.equal(offer.delegatedCompletion?.messageId, steerMessageId); + 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", @@ -495,17 +513,17 @@ it.effect.each([ if (!resultRead) { const wakeA = yield* nextWake(1, { taskIds: [taskA] }); assert.equal(wakeA.message.messageId, steerMessageId); - yield* settle(wakeA.runId, wakeA.runOrdinal); + yield* settle(wakeA); } if (queuedMessage) { const queued = yield* nextWake(1, { taskIds: [] }); assert.equal(queued.message.messageId, queuedMessageId); - yield* settle(queued.runId, queued.runOrdinal); + yield* settle(queued); } const wakeB = yield* nextWake(resultRead ? (queuedMessage ? 2 : 1) : 2, { taskIds: [taskB], }); - yield* settle(wakeB.runId, wakeB.runOrdinal); + yield* settle(wakeB); const final = yield* orchestrator.getThreadProjection(threadId); assert.equal(final.runs[0]?.delegatedCompletion?.delivery, null);