From b206d6de1ddb6fe812eadaf7b1c2bfe9e8fa3857 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 9 Nov 2026 16:55:00 +0000 Subject: [PATCH 1/3] test(server): reproduce #33454 steered completion wake strands its cohort A steer that restarts a delegated-completion wake run leaves the generation's tasks claimed and never wakes the parent for a sibling that finishes afterwards. These tests reproduce #33454. --- .../SteeringCompletion.integration.test.ts | 224 ++++++++++++++++++ 1 file changed, 224 insertions(+) diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 2c311f2569753..09c4f1bac69f8 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -2,6 +2,7 @@ import { assert, it } from "@effect/vitest"; import { ChatAttachmentId, CommandId, + EnvironmentId, MessageId, EventId, NodeId, @@ -1215,6 +1216,229 @@ it.effect("queues a steer until the turn ends when the provider cannot steer it" ), ); +// regression: #33454 a steer that restarts a completion wake run left its cohort claimed forever +it.effect("wakes the parent for a sibling result after a steer restarts the completion wake", () => + Effect.scoped( + Effect.gen(function* () { + const { events, started, layer, startFirstTurn } = yield* nextTurnSelectionHarness( + "steered-completion-wake", + true, + false, + ); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* startFirstTurn; + const watchRun = (runId: string, status: string) => + orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === status, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const watchTurnRunning = (ordinal: number) => + orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.ordinal === ordinal, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const finishTurn = (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + const providerTurn = (yield* orchestrator.getThreadProjection( + threadId, + )).providerTurns.find((candidate) => candidate.runAttemptId === turn.attemptId)!; + const settled = yield* watchRun(turn.runId, "waiting"); + yield* Queue.offer(events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...providerTurn, + status: "completed", + completedAt: yield* DateTime.now, + }, + }); + yield* Queue.offer(events, { + type: "turn.terminal", + driver, + providerThreadId: providerTurn.providerThreadId, + providerTurnId: providerTurn.id, + runOrdinal: turn.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + }); + + const origin = started[0]!; + const firstTaskId = NodeId.make("task:steered-wake-first"); + const siblingTaskId = NodeId.make("task:steered-wake-sibling"); + const now = yield* DateTime.now; + yield* sink.write({ + events: [firstTaskId, siblingTaskId].map((taskId) => ({ + id: EventId.make(`event:seed:${taskId}`), + type: "subagent.updated" as const, + threadId, + runId: origin.runId, + nodeId: taskId, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId: origin.runId, + parentNodeId: origin.rootNodeId, + origin: "app_owned" as const, + createdBy: "agent" as const, + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: null, + remoteChild: { + environmentId: EnvironmentId.make("environment:box"), + threadId: ThreadId.make(`thread:remote:${taskId}`), + label: "box", + }, + nativeTaskRef: null, + prompt: "Do background work", + title: null, + model: null, + completionWake: "always" as const, + completionDelivery: { state: "pending" as const, observedByRunId: null }, + status: "running" as const, + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + })), + }); + yield* finishTurn(origin); + + const completeTask = (taskId: NodeId) => + orchestrator.dispatch({ + type: "delegated_task.remote.complete", + commandId: CommandId.make(`command:remote-complete:${taskId}`), + parentThreadId: threadId, + taskId, + status: "completed", + result: `${taskId} finished`, + stoppedThere: "ran-out", + }); + yield* completeTask(firstTaskId); + const firstDelivery = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === origin.runId, + )!.delegatedCompletion!.delivery!; + assert.deepEqual(firstDelivery.taskIds, [firstTaskId]); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:first-wake"), + threadId, + messageId: firstDelivery.messageId, + text: "Delegated task completed", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { + parentRunId: origin.runId, + generation: firstDelivery.generation, + taskIds: firstDelivery.taskIds, + }, + createdBy: "agent", + creationSource: "server", + }); + const wakeRunning = yield* watchTurnRunning(2); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + yield* Fiber.join(wakeRunning); + const wake = started[1]!; + assert.equal(wake.message.messageId, firstDelivery.messageId); + + // The provider cannot steer natively, so the steer restarts the wake run. + const steeredRunning = yield* watchTurnRunning(3); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:steer-wake"), + threadId, + messageId: MessageId.make("message:steer-wake"), + text: "also check the logs", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: wake.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(steeredRunning); + const steered = started[2]!; + assert.equal(steered.runId, wake.runId); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).attempts.find( + (attempt) => attempt.id === steered.attemptId, + )?.reason, + "steering_restart", + ); + + // A follow-up queued behind the wake starts once the terminal-run listener has settled it. + const followUp = MessageId.make("message:follow-up"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:follow-up"), + threadId, + messageId: followUp, + text: "follow up", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + const followUpRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.userMessageId === followUp, + )!; + const followUpStarted = yield* watchRun(followUpRun.id, "starting"); + yield* finishTurn(steered); + yield* Fiber.join(followUpStarted); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === wake.runId, + )?.status, + "completed", + ); + + yield* completeTask(siblingTaskId); + const final = yield* orchestrator.getThreadProjection(threadId); + const tasks = new Map( + final.subagents.map((task) => [task.id, task.completionDelivery?.state]), + ); + assert.deepInclude( + final.runs.find((run) => run.id === origin.runId)?.delegatedCompletion?.delivery ?? {}, + { generation: 2, taskIds: [siblingTaskId] }, + "the sibling finished after the steered wake but no new completion wake was reserved", + ); + assert.equal(tasks.get(siblingTaskId), "claimed"); + assert.equal( + tasks.get(firstTaskId), + "delivered", + "the steered wake run's own task stays claimed after that run completed", + ); + }).pipe(Effect.provide(layer)); + }), + ), +); + // Codex flushes held drafts after the final answer; reserve their pre-steer position before turn end. it.effect("keeps a held commentary draft before the steer when turn end flushes it", () => Effect.scoped( From d78c9c10a1db80318f8fa2ceda2cea363baef071 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 9 Nov 2026 17:05:39 +0000 Subject: [PATCH 2/3] fix(server): a steer that restarts a completion wake settles its cohort A steer that restarts a delegated-completion wake run replaces the run's userMessageId, so the wake's terminal could no longer find its delivery and the cohort stayed claimed: sibling results that finished later never woke the parent again. The restart now settles the wake the provider already received, using the same logic as an accepted mailbox delivery (extracted into acceptDelegatedCompletionDelivery), and offers any next generation it reserves once the steer commits. Fixes pingdotgg/t3code#33454 --- .../src/orchestration-v2/Orchestrator.ts | 214 +++++---- .../SteeringCompletion.integration.test.ts | 410 +++++++++++------- 2 files changed, 373 insertions(+), 251 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 37c4a75f663b6..283118891b120 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1581,6 +1581,104 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } }); + /** Marks a wake the provider received delivered and reserves the next one for pending siblings. */ + const acceptDelegatedCompletionDelivery = Effect.fn( + "Orchestrator.acceptDelegatedCompletionDelivery", + )(function* (input: { + readonly command: OrchestrationV2ServerCommand; + readonly events: Ref.Ref>; + readonly projection: Pick< + OrchestrationV2ThreadProjection, + "thread" | "runs" | "messages" | "subagents" + >; + readonly messageId: OrchestrationV2ConversationMessage["id"]; + readonly now: DateTime.Utc; + }) { + const { projection, now } = input; + const message = projection.messages.find((row) => row.id === input.messageId); + const ownership = message?.delegatedCompletion; + const parentRun = projection.runs.find((run) => run.id === ownership?.parentRunId); + const cohort = parentRun?.delegatedCompletion; + const delivery = cohort?.delivery; + if ( + parentRun === undefined || + cohort === undefined || + cohort.disposition !== "open" || + delivery == null || + delivery.messageId !== input.messageId || + delivery.generation !== ownership?.generation + ) { + return undefined; + } + const threadId = projection.thread.id; + 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 emitEvent = emit(input.events, input.command); + 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, + runId: parentRun.id, + nodeId: task.id, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + occurredAt: now, + payload: { ...task, completionDelivery: { state, observedByRunId: null }, updatedAt: now }, + }); + } + yield* emitEvent({ + 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 parentRun.id; + }); + const emitQueuedRunCancellation = Effect.fn("Orchestrator.emitQueuedRunCancellation")( function* (input: { readonly command: OrchestrationV2ServerCommand; @@ -5673,6 +5771,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: checkpointScope, }); + // The restart replaces the run's input, so the wake it was running can no longer be + // settled at its terminal. Its provider turn was running, so the wake already landed. + yield* acceptDelegatedCompletionDelivery({ + command: input.command, + events: input.events, + projection: input.projection, + messageId: targetRun.userMessageId, + now, + }); yield* appendSteeringMessage({ runId: targetRun.id, nodeId: nextRootNodeId, @@ -13308,95 +13415,26 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: Ref.Ref>, ) { const projection = yield* getProjectionWithPendingEvents(command.threadId, events); - const message = projection.messages.find((row) => row.id === command.messageId); - const ownership = message?.delegatedCompletion; - const parentRun = projection.runs.find((run) => run.id === ownership?.parentRunId); - const cohort = parentRun?.delegatedCompletion; - const delivery = cohort?.delivery; const now = yield* DateTime.now; - const emitEvent = emit(events, command); - if ( - parentRun === undefined || - cohort === undefined || - cohort.disposition !== "open" || - delivery == null || - delivery.messageId !== command.messageId || - delivery.generation !== ownership?.generation - ) { - yield* emitEvent({ + const parentRunId = yield* acceptDelegatedCompletionDelivery({ + command, + events, + projection, + messageId: command.messageId, + now, + }); + if (parentRunId === undefined) { + yield* emit( + events, + command, + )({ type: "thread.metadata-updated", threadId: command.threadId, occurredAt: now, payload: projection.thread, }); - return; } - 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 }, - }); - } - 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, - }, - }, - }); - return parentRun.id; + return parentRunId; }); const dispatchUnsupported = (command: OrchestrationV2ServerCommand) => @@ -14086,7 +14124,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ), ); } - if (command.type === "notification.delivery.accept") { + if ( + command.type === "notification.delivery.accept" || + // A steer restart that settled a wake can reserve the next one for its siblings. + ((command.type === "message.dispatch" || + command.type === "queued-message.promote-to-steer") && + committed.storedEvents.some( + ({ event }) => + event.type === "subagent.updated" && + event.payload.completionDelivery?.state === "claimed", + )) + ) { yield* mapDispatchError(command)(offerDelegatedCompletionDeliveries(command.threadId)); } if ( diff --git a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts index 09c4f1bac69f8..eb4ebd0363641 100644 --- a/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts +++ b/apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts @@ -20,6 +20,7 @@ import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; @@ -30,6 +31,7 @@ import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; const driver = ProviderDriverKind.make("codex"); const instanceId = ProviderInstanceId.make("codex"); @@ -1217,179 +1219,211 @@ it.effect("queues a steer until the turn ends when the provider cannot steer it" ); // regression: #33454 a steer that restarts a completion wake run left its cohort claimed forever +const restartCompletionWake = Effect.fn("restartCompletionWake")(function* ( + harness: Pick< + Effect.Success>, + "events" | "started" | "startFirstTurn" + >, + siblingFinishesFirst: boolean, +) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const sink = yield* EventSink.EventSinkV2; + const threadId = yield* harness.startFirstTurn; + const watchRun = (runId: string, status: string) => + orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "run.updated" && + event.payload.id === runId && + event.payload.status === status, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const watchTurnRunning = (ordinal: number) => + orchestrator.streamDomainEvents.pipe( + Stream.filter( + (event) => + event.type === "provider-turn.updated" && + event.payload.status === "running" && + event.payload.ordinal === ordinal, + ), + Stream.take(1), + Stream.runDrain, + Effect.forkScoped, + ); + const finishTurn = (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.gen(function* () { + const providerTurn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns.find( + (candidate) => candidate.runAttemptId === turn.attemptId, + )!; + const settled = yield* watchRun(turn.runId, "waiting"); + yield* Queue.offer(harness.events, { + type: "provider_turn.updated", + driver, + providerTurn: { + ...providerTurn, + status: "completed", + completedAt: yield* DateTime.now, + }, + }); + yield* Queue.offer(harness.events, { + type: "turn.terminal", + driver, + providerThreadId: providerTurn.providerThreadId, + providerTurnId: providerTurn.id, + runOrdinal: turn.runOrdinal, + status: "completed", + failure: null, + threadDisposition: "reusable", + }); + yield* Fiber.join(settled); + yield* worker.drain(); + }); + + const origin = harness.started[0]!; + const firstTaskId = NodeId.make("task:steered-wake-first"); + const siblingTaskId = NodeId.make("task:steered-wake-sibling"); + const now = yield* DateTime.now; + yield* sink.write({ + events: [firstTaskId, siblingTaskId].map((taskId) => ({ + id: EventId.make(`event:seed:${taskId}`), + type: "subagent.updated" as const, + threadId, + runId: origin.runId, + nodeId: taskId, + occurredAt: now, + payload: { + id: taskId, + threadId, + runId: origin.runId, + parentNodeId: origin.rootNodeId, + origin: "app_owned" as const, + createdBy: "agent" as const, + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: null, + remoteChild: { + environmentId: EnvironmentId.make("environment:box"), + threadId: ThreadId.make(`thread:remote:${taskId}`), + label: "box", + }, + nativeTaskRef: null, + prompt: "Do background work", + title: null, + model: null, + completionWake: "always" as const, + completionDelivery: { state: "pending" as const, observedByRunId: null }, + status: "running" as const, + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + })), + }); + yield* finishTurn(origin); + + const completeTask = (taskId: NodeId) => + orchestrator.dispatch({ + type: "delegated_task.remote.complete", + commandId: CommandId.make(`command:remote-complete:${taskId}`), + parentThreadId: threadId, + taskId, + status: "completed", + result: `${taskId} finished`, + stoppedThere: "ran-out", + }); + yield* completeTask(firstTaskId); + const firstDelivery = (yield* orchestrator.getThreadProjection(threadId)).runs.find( + (run) => run.id === origin.runId, + )!.delegatedCompletion!.delivery!; + assert.deepEqual(firstDelivery.taskIds, [firstTaskId]); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:first-wake"), + threadId, + messageId: firstDelivery.messageId, + text: "Delegated task completed", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "queue_after_active" }, + delegatedCompletion: { + parentRunId: origin.runId, + generation: firstDelivery.generation, + taskIds: firstDelivery.taskIds, + }, + createdBy: "agent", + creationSource: "server", + }); + const wakeRunning = yield* watchTurnRunning(2); + yield* orchestrator.resumeQueuedRuns; + yield* worker.drain(); + yield* Fiber.join(wakeRunning); + const wake = harness.started[1]!; + assert.equal(wake.message.messageId, firstDelivery.messageId); + + if (siblingFinishesFirst) { + yield* completeTask(siblingTaskId); + } + + // The provider cannot steer natively, so the steer restarts the wake run. + const steeredRunning = yield* watchTurnRunning(3); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:steer-wake"), + threadId, + messageId: MessageId.make("message:steer-wake"), + text: "also check the logs", + attachments: [], + modelSelection: runSelection, + dispatchMode: { type: "steer_active", targetRunId: wake.runId }, + createdBy: "user", + creationSource: "web", + }); + yield* worker.drain(); + yield* Fiber.join(steeredRunning); + const steered = harness.started[2]!; + assert.equal(steered.runId, wake.runId); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).attempts.find( + (attempt) => attempt.id === steered.attemptId, + )?.reason, + "steering_restart", + ); + return { + threadId, + origin, + wake, + steered, + firstTaskId, + siblingTaskId, + watchRun, + finishTurn, + completeTask, + }; +}); + it.effect("wakes the parent for a sibling result after a steer restarts the completion wake", () => Effect.scoped( Effect.gen(function* () { - const { events, started, layer, startFirstTurn } = yield* nextTurnSelectionHarness( - "steered-completion-wake", - true, - false, - ); + const harness = yield* nextTurnSelectionHarness("steered-completion-wake", true, false); yield* Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; - const sink = yield* EventSink.EventSinkV2; - const threadId = yield* startFirstTurn; - const watchRun = (runId: string, status: string) => - orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "run.updated" && - event.payload.id === runId && - event.payload.status === status, - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - const watchTurnRunning = (ordinal: number) => - orchestrator.streamDomainEvents.pipe( - Stream.filter( - (event) => - event.type === "provider-turn.updated" && - event.payload.status === "running" && - event.payload.ordinal === ordinal, - ), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped, - ); - const finishTurn = (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => - Effect.gen(function* () { - const providerTurn = (yield* orchestrator.getThreadProjection( - threadId, - )).providerTurns.find((candidate) => candidate.runAttemptId === turn.attemptId)!; - const settled = yield* watchRun(turn.runId, "waiting"); - yield* Queue.offer(events, { - type: "provider_turn.updated", - driver, - providerTurn: { - ...providerTurn, - status: "completed", - completedAt: yield* DateTime.now, - }, - }); - yield* Queue.offer(events, { - type: "turn.terminal", - driver, - providerThreadId: providerTurn.providerThreadId, - providerTurnId: providerTurn.id, - runOrdinal: turn.runOrdinal, - status: "completed", - failure: null, - threadDisposition: "reusable", - }); - yield* Fiber.join(settled); - yield* worker.drain(); - }); - - const origin = started[0]!; - const firstTaskId = NodeId.make("task:steered-wake-first"); - const siblingTaskId = NodeId.make("task:steered-wake-sibling"); - const now = yield* DateTime.now; - yield* sink.write({ - events: [firstTaskId, siblingTaskId].map((taskId) => ({ - id: EventId.make(`event:seed:${taskId}`), - type: "subagent.updated" as const, - threadId, - runId: origin.runId, - nodeId: taskId, - occurredAt: now, - payload: { - id: taskId, - threadId, - runId: origin.runId, - parentNodeId: origin.rootNodeId, - origin: "app_owned" as const, - createdBy: "agent" as const, - driver, - providerInstanceId: instanceId, - providerThreadId: null, - childThreadId: null, - remoteChild: { - environmentId: EnvironmentId.make("environment:box"), - threadId: ThreadId.make(`thread:remote:${taskId}`), - label: "box", - }, - nativeTaskRef: null, - prompt: "Do background work", - title: null, - model: null, - completionWake: "always" as const, - completionDelivery: { state: "pending" as const, observedByRunId: null }, - status: "running" as const, - result: null, - startedAt: now, - completedAt: null, - updatedAt: now, - }, - })), - }); - yield* finishTurn(origin); - - const completeTask = (taskId: NodeId) => - orchestrator.dispatch({ - type: "delegated_task.remote.complete", - commandId: CommandId.make(`command:remote-complete:${taskId}`), - parentThreadId: threadId, - taskId, - status: "completed", - result: `${taskId} finished`, - stoppedThere: "ran-out", - }); - yield* completeTask(firstTaskId); - const firstDelivery = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (run) => run.id === origin.runId, - )!.delegatedCompletion!.delivery!; - assert.deepEqual(firstDelivery.taskIds, [firstTaskId]); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("command:first-wake"), - threadId, - messageId: firstDelivery.messageId, - text: "Delegated task completed", - attachments: [], - modelSelection: runSelection, - dispatchMode: { type: "queue_after_active" }, - delegatedCompletion: { - parentRunId: origin.runId, - generation: firstDelivery.generation, - taskIds: firstDelivery.taskIds, - }, - createdBy: "agent", - creationSource: "server", - }); - const wakeRunning = yield* watchTurnRunning(2); - yield* orchestrator.resumeQueuedRuns; - yield* worker.drain(); - yield* Fiber.join(wakeRunning); - const wake = started[1]!; - assert.equal(wake.message.messageId, firstDelivery.messageId); - - // The provider cannot steer natively, so the steer restarts the wake run. - const steeredRunning = yield* watchTurnRunning(3); - yield* orchestrator.dispatch({ - type: "message.dispatch", - commandId: CommandId.make("command:steer-wake"), + const { threadId, - messageId: MessageId.make("message:steer-wake"), - text: "also check the logs", - attachments: [], - modelSelection: runSelection, - dispatchMode: { type: "steer_active", targetRunId: wake.runId }, - createdBy: "user", - creationSource: "web", - }); - yield* worker.drain(); - yield* Fiber.join(steeredRunning); - const steered = started[2]!; - assert.equal(steered.runId, wake.runId); - assert.equal( - (yield* orchestrator.getThreadProjection(threadId)).attempts.find( - (attempt) => attempt.id === steered.attemptId, - )?.reason, - "steering_restart", - ); + origin, + wake, + steered, + firstTaskId, + siblingTaskId, + watchRun, + finishTurn, + completeTask, + } = yield* restartCompletionWake(harness, false); // A follow-up queued behind the wake starts once the terminal-run listener has settled it. const followUp = MessageId.make("message:follow-up"); @@ -1434,11 +1468,51 @@ it.effect("wakes the parent for a sibling result after a steer restarts the comp "delivered", "the steered wake run's own task stays claimed after that run completed", ); - }).pipe(Effect.provide(layer)); + }).pipe(Effect.provide(harness.layer)); }), ), ); +it.effect( + "offers the next completion wake when a steer restarts a wake a sibling finished behind", + () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* nextTurnSelectionHarness( + "steered-completion-wake-behind", + true, + false, + ); + const offers: ProviderContinuationRequests.ProviderContinuationRequest[] = []; + const requests = Layer.succeed(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: (request) => Effect.sync(() => void offers.push(request)), + take: Effect.never, + }); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { threadId, origin, firstTaskId, siblingTaskId } = yield* restartCompletionWake( + harness, + true, + ); + const projection = yield* orchestrator.getThreadProjection(threadId); + const delivery = projection.runs.find((run) => run.id === origin.runId) + ?.delegatedCompletion?.delivery; + assert.deepInclude(delivery ?? {}, { generation: 2, taskIds: [siblingTaskId] }); + const tasks = new Map( + projection.subagents.map((task) => [task.id, task.completionDelivery?.state]), + ); + assert.equal(tasks.get(firstTaskId), "delivered"); + assert.equal(tasks.get(siblingTaskId), "claimed"); + assert.deepEqual(offers.at(-1)?.delegatedCompletion, { + parentRunId: origin.runId, + generation: 2, + messageId: delivery!.messageId, + }); + }).pipe(Effect.provide(harness.layer.pipe(Layer.provide(requests)))); + }), + ), +); + // Codex flushes held drafts after the final answer; reserve their pre-steer position before turn end. it.effect("keeps a held commentary draft before the steer when turn end flushes it", () => Effect.scoped( From 87ae360b64b994c4dee7de1670c31615bf11de76 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 02:37:53 +0000 Subject: [PATCH 3/3] chore: retrigger branch sync Co-Authored-By: Claude Opus 5.5