From 265fb7000f84d12a44c4e87c17426e1f29e0bb36 Mon Sep 17 00:00:00 2001 From: Yash Singh Date: Mon, 12 Oct 2026 20:26:16 -0500 Subject: [PATCH] fix(server): apply confirmed Claude subagent outcomes (#18778) --- .../Adapters/ClaudeAdapterV2.test.ts | 430 +++++++++++++++++- .../Adapters/ClaudeAdapterV2.ts | 160 ++++++- .../src/orchestration-v2/EffectWorker.test.ts | 90 +++- .../src/orchestration-v2/EffectWorker.ts | 56 ++- .../src/orchestration-v2/Orchestrator.ts | 56 ++- .../ProviderTurnControlService.ts | 31 +- .../src/orchestration-v2/ThreadStop.test.ts | 175 ++++++- packages/contracts/src/orchestrationV2.ts | 15 +- .../src/server/ProviderAdapter.ts | 11 +- .../src/server/subagentProjection.ts | 39 +- 10 files changed, 979 insertions(+), 84 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 7460977ecb1b6fa47f5ab34e29e7387a803a5445..55df305dc00e40a3b10690c5657545651446d290 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -60,6 +60,7 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; +import * as Scheduler from "effect/Scheduler"; import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; @@ -2317,7 +2318,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { const makeWakeHarnessWithOptions = (options?: { readonly close?: (sdkMessages: Queue.Queue) => Effect.Effect; readonly interrupt?: Effect.Effect; - readonly stopTask?: (taskId: string) => Effect.Effect; + readonly stopTask?: ( + taskId: string, + ) => Effect.Effect; readonly environment?: NodeJS.ProcessEnv; readonly subagentLaunchToolUseId?: string; readonly nativeSessionId?: string; @@ -8792,17 +8795,373 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); it.effect.each([false, true])( - "stops one native subagent while its sibling keeps running, owner settled %s", - (ownerSettled) => + "drops an idle start superseded by live completion, later resume %s", + (resume) => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const startTurn = (ordinal: number, continuation = false) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: DateTime.makeUnsafe(0), + attemptId: RunAttemptId.make(`attempt-stale-start-${ordinal}`), + providerTurnOrdinal: ordinal, + text: "Continue work.", + attachments: [], + ...(continuation + ? { + messageCreatedBy: "agent" as const, + messageCreationSource: "provider" as const, + } + : {}), + }), + ); + const started = (taskId: string, suffix: string) => + makeSubagentTaskStartedFrame({ + taskId, + toolUseId: `toolu-${taskId}`, + uuid: `00000000-0000-4000-8000-${suffix}`, + }); + const finished = (taskId: string, suffix: string) => + makeSubagentNotificationFrame({ + taskId, + toolUseId: `toolu-${taskId}`, + summary: "Finished audit.", + uuid: `00000000-0000-4000-8000-${suffix}`, + }); + yield* startTurn(1); + yield* harness.offerAndWait(started("stale-agent", "000000004001")); + yield* harness.offerAndWait(started("wake-agent", "000000004002")); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000004003", + result: "Agents launched.", + }), + ); + yield* Queue.take(harness.terminalReceipts); + // The background task resumes in the idle gap, before a user turn. + yield* harness.offerAndWait(started("stale-agent", "000000004004")); + yield* startTurn(2); + yield* harness.offerAndWait(finished("stale-agent", "000000004005")); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000004006", + result: "User turn finished.", + }), + ); + yield* Queue.take(harness.terminalReceipts); + if (resume) yield* harness.offerAndWait(started("stale-agent", "000000004007")); + yield* harness.offerAndWait(finished("wake-agent", "000000004008")); + yield* Queue.take(harness.continuationReceipts); + yield* startTurn(3, true); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000004009", + result: "Wake finished.", + }), + ); + yield* Queue.take(harness.terminalReceipts); + const latest = harness.events.findLast( + (event) => + event.type === "subagent.updated" && + event.subagent.nativeTaskRef?.nativeId === "stale-agent", + ); + assert.equal(latest?.type, "subagent.updated"); + if (latest?.type !== "subagent.updated") return; + assert.equal(latest.subagent.status, resume ? "running" : "completed"); + assert.equal(latest.subagent.result, resume ? null : "Finished audit."); + assert.equal(latest.subagent.completedAt === null, resume); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + const makeStopFixture = (options?: Parameters[0]) => + Effect.gen(function* () { + const requests = yield* Queue.unbounded(); + const harness = yield* makeWakeHarnessWithOptions({ + ...options, + stopTask: (id) => + Queue.offer(requests, undefined).pipe( + Effect.andThen(options?.stopTask?.(id) ?? Effect.void), + ), + }); + let sequence = 4500; + const uuid = () => `00000000-0000-4000-8000-${String(++sequence).padStart(12, "0")}`; + const startTurn = (continuation = false) => + harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: DateTime.makeUnsafe(0), + attemptId: RunAttemptId.make(`attempt-stop-${continuation ? 2 : 1}`), + providerTurnOrdinal: continuation ? 2 : 1, + text: "Run an audit.", + attachments: [], + ...(continuation + ? { messageCreatedBy: "agent" as const, messageCreationSource: "provider" as const } + : {}), + }), + ); + const start = (toolUseId = "toolu-stop-agent") => + harness.offerAndWait( + makeSubagentTaskStartedFrame({ + taskId: "stop-agent", + toolUseId, + uuid: uuid(), + }), + ); + const notify = ( + status: "completed" | "failed" | "stopped", + summary = "Native outcome.", + toolUseId = "toolu-stop-agent", + ) => + harness.offerAndWait( + claudeSdkFrame({ + ...makeSubagentNotificationFrame({ + taskId: "stop-agent", + toolUseId, + summary, + uuid: uuid(), + }), + status, + }), + ); + const stop = (requestedAt?: DateTime.Utc) => + harness.runtime.stopSubagent!({ + providerThread: harness.providerThread, + nativeTaskId: "stop-agent", + ...(requestedAt === undefined ? {} : { requestedAt }), + }); + const finishParent = () => + harness + .offerAndWait(makeResultFrame({ uuid: uuid(), result: "Launched." })) + .pipe(Effect.andThen(Queue.take(harness.terminalReceipts))); + yield* startTurn(); + yield* start(); + yield* Queue.take(harness.subagentReceipts); + return { + ...harness, + start, + notify, + stop, + finishParent, + replay: () => startTurn(true), + uuid, + requests, + }; + }); + + it.effect.each([ + [false, "no notification"], + [true, "no notification"], + [false, "stop fails"], + [false, "stop hangs"], + ] as const)("never infers an outcome from Stop, owner settled %s and %s", ([idle, failure]) => + Effect.scoped( + Effect.gen(function* () { + const requested = yield* Deferred.make(); + const fixture = yield* makeStopFixture({ + stopTask: () => + Deferred.succeed(requested, undefined).pipe( + Effect.andThen( + failure === "stop fails" + ? new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({ + method: "stopTask", + cause: "Task is no longer live", + }) + : failure === "stop hangs" + ? Effect.never + : Effect.void, + ), + ), + }); + if (idle) yield* fixture.finishParent(); + const stopping = yield* fixture.stop().pipe(Effect.exit, Effect.forkScoped); + yield* Deferred.await(requested); + yield* TestClock.adjust("2 seconds"); + const result = yield* Fiber.join(stopping); + assert.isTrue(Exit.isFailure(result)); + assert.equal( + fixture.events.findLast((event) => event.type === "subagent.updated")?.subagent.status, + "running", + ); + assert.isTrue(yield* fixture.hasPendingBackgroundWork); + yield* fixture.notify("completed"); + assert.equal((yield* Queue.take(fixture.subagentReceipts)).subagent.status, "completed"); + assert.isFalse(yield* fixture.hasPendingBackgroundWork); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect.each([1, 2, 3, 4, 5, 6, 7, 8])( + "accepts completion during Stop at scheduler step %s", + (pauseAt) => + Effect.scoped( + Effect.gen(function* () { + const fixture = yield* makeStopFixture(); + const resumptions = yield* Queue.unbounded<() => void>(); + const scheduler = yield* Scheduler.Scheduler; + const dispatcher = scheduler.makeDispatcher(); + let checks = 0; + // Pause Stop before its SDK request and deliver the outcome while it is suspended. + const stopping = yield* fixture.stop().pipe( + Effect.provideService(Scheduler.Scheduler, { + executionMode: scheduler.executionMode, + shouldYield: () => ++checks === pauseAt, + makeDispatcher: () => ({ + scheduleTask: (task, priority) => { + if (checks === pauseAt) Queue.offerUnsafe(resumptions, task); + else dispatcher.scheduleTask(task, priority); + }, + flush: () => dispatcher.flush(), + }), + }), + Effect.exit, + Effect.forkScoped, + ); + const resume = yield* Queue.take(resumptions); + yield* fixture.notify("completed"); + const completed = yield* Queue.take(fixture.subagentReceipts); + yield* Effect.sync(resume); + yield* TestClock.adjust("2 seconds"); + assert.deepEqual( + yield* Fiber.join(stopping), + Exit.succeed({ + status: "completed" as const, + completedAt: completed.subagent.completedAt!, + result: "Native outcome.", + }), + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect("retains native confirmation after the synchronous Agent tool result", () => + Effect.scoped( + Effect.gen(function* () { + const fixture = yield* makeStopFixture({ + stopTask: () => Effect.die("Task already completed"), + }); + yield* fixture.notify("completed"); + const completed = yield* Queue.take(fixture.subagentReceipts); + yield* fixture.offerAndWait( + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { type: "tool_result", tool_use_id: "toolu-stop-agent", content: "Tool result." }, + ], + }, + parent_tool_use_id: null, + uuid: fixture.uuid(), + session_id: WAKE_NATIVE_SESSION, + }), + ); + assert.equal((yield* Queue.take(fixture.subagentReceipts)).subagent.result, "Tool result."); + assert.deepEqual(yield* fixture.stop(), { + status: "completed", + completedAt: completed.subagent.completedAt!, + result: "Native outcome.", + }); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect.each(["completed", "failed", "stopped"] as const)( + "applies idle %s once, preserves a genuine resume, and only confirms native outcomes", + (status) => + Effect.scoped( + Effect.gen(function* () { + let stops = 0; + const fixture = yield* makeStopFixture({ + stopTask: () => Effect.sync(() => void stops++), + }); + yield* fixture.finishParent(); + yield* TestClock.adjust("1 second"); + yield* fixture.notify(status); + const completed = yield* Queue.take(fixture.subagentReceipts); + const expected = status === "stopped" ? "cancelled" : status; + assert.equal(completed.subagent.status, expected); + assert.lengthOf(fixture.offeredMessages, 1); + assert.isFalse(yield* fixture.hasPendingBackgroundWork); + assert.deepEqual(yield* fixture.stop(), { + status: expected, + completedAt: completed.subagent.completedAt!, + result: "Native outcome.", + }); + assert.equal(stops, 0); + yield* TestClock.adjust("1 second"); + yield* fixture.start("toolu-resume"); + assert.isUndefined(yield* fixture.stop(completed.subagent.completedAt!)); + assert.equal(stops, 0); + const stopping = yield* fixture.stop().pipe(Effect.exit, Effect.forkScoped); + yield* Queue.take(fixture.requests); + const beforeReplay = fixture.events.length; + yield* fixture.replay(); + const reopened = yield* Queue.take(fixture.subagentReceipts); + assert.equal(reopened.subagent.status, "running"); + assert.isNull(reopened.subagent.completedAt); + assert.isFalse( + fixture.events + .slice(beforeReplay) + .some( + (event) => event.type === "subagent.updated" && event.subagent.status !== "running", + ), + ); + yield* TestClock.adjust("2 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(stopping))); + assert.equal(stops, 1); + yield* fixture.notify("completed", "Resumed outcome.", "toolu-resume"); + const resumed = yield* Queue.take(fixture.subagentReceipts); + assert.equal(resumed.subagent.result, "Resumed outcome."); + assert.isAbove( + DateTime.toEpochMillis(resumed.subagent.completedAt!), + DateTime.toEpochMillis(completed.subagent.completedAt!), + ); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + + it.effect.each([ + [false, "stopped"], + [true, "stopped"], + [false, "completed"], + ] as const)( + "stops one native subagent while its sibling keeps running, owner settled %s, outcome %s", + ([ownerSettled, outcome]) => Effect.scoped( Effect.gen(function* () { const stopped: string[] = []; + const stopRequested = yield* Deferred.make(); let closes = 0; const harness = yield* makeWakeHarnessWithOptions({ stopTask: (taskId) => Effect.sync(() => { stopped.push(taskId); - }), + }).pipe(Effect.andThen(Deferred.succeed(stopRequested, undefined))), interrupt: Effect.die("A child stop must not interrupt its owner"), close: () => Effect.sync(() => { @@ -8839,7 +9198,11 @@ describe("ClaudeAdapterV2 background wake turns", () => { } const stop = harness.runtime.stopSubagent; assert.isDefined(stop); - yield* stop!({ providerThread: harness.providerThread, nativeTaskId: "agent-stop" }); + const stopping = yield* stop!({ + providerThread: harness.providerThread, + nativeTaskId: "agent-stop", + }).pipe(Effect.forkScoped); + yield* Deferred.await(stopRequested); assert.deepEqual(stopped, ["agent-stop"]); yield* harness.offerAndWait( claudeSdkFrame({ @@ -8849,7 +9212,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { summary: "Stopped by user", uuid: "00000000-0000-4000-8000-000000000804", }), - status: "stopped", + status: outcome, }), ); if (ownerSettled) { @@ -8867,13 +9230,17 @@ describe("ClaudeAdapterV2 background wake turns", () => { }), ); } + yield* Fiber.join(stopping); const stoppedEvent = yield* Queue.take(harness.subagentReceipts); - assert.equal(stoppedEvent.subagent.status, "cancelled"); + assert.equal( + stoppedEvent.subagent.status, + outcome === "stopped" ? "cancelled" : "completed", + ); const tasks = harness.events.filter((event) => event.type === "subagent.updated"); assert.equal( tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-stop") ?.subagent.status, - "cancelled", + outcome === "stopped" ? "cancelled" : "completed", ); assert.equal( tasks.findLast((event) => event.subagent.nativeTaskRef?.nativeId === "agent-keep") @@ -10404,7 +10771,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { () => events.filter((event) => event.type === "turn.terminal").length === 2, "user turn terminal after replace", ); - assert.isTrue(yield* hasPendingBackgroundWork); + // The actual completion was applied while idle, before wake replay. + assert.isFalse(yield* hasPendingBackgroundWork); + assert.equal(subagentEvents().at(-1)?.subagent.status, "completed"); assert.lengthOf(continuationRequests, 1); yield* Queue.offer( @@ -12700,17 +13069,24 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); it.effect.each([ - ["api_error", false], - ["aborted_streaming", false], - ["api_error", true], + ["api_error", false, false], + ["aborted_streaming", false, false], + ["api_error", true, false], + ["api_error", false, true], + ["aborted_streaming", false, true], ] as const)( - "stops the run's live workflow after root result %s, continuation=%s", - ([terminalReason, continuation]) => + "stops the run's live workflow after root result %s, continuation=%s, pending Stop=%s", + ([terminalReason, continuation, pendingStop]) => Effect.scoped( Effect.gen(function* () { const stoppedTasks: Array = []; + const stopRequested = yield* Deferred.make(); const harness = yield* makeWorkflowHarnessWithOptions({ - stopTask: (taskId) => Effect.sync(() => void stoppedTasks.push(taskId)), + stopTask: (taskId) => + Effect.sync(() => void stoppedTasks.push(taskId)).pipe( + Effect.andThen(Deferred.succeed(stopRequested, undefined)), + Effect.asVoid, + ), }); const now = yield* DateTime.now; const turnInput = { @@ -12723,6 +13099,13 @@ describe("ClaudeAdapterV2 background wake turns", () => { }; yield* harness.runtime.startTurn(makeClaudeTestTurnInput(turnInput)); yield* launchWorkflow({ harness, snapshotUuid: "00000000-0000-4000-8000-000000001050" }); + const stopping = pendingStop + ? yield* harness.runtime.stopSubagent!({ + providerThread: harness.providerThread, + nativeTaskId: WORKFLOW_TASK_ID, + }).pipe(Effect.exit, Effect.forkScoped) + : undefined; + if (pendingStop) yield* Deferred.await(stopRequested); if (continuation) { yield* harness.offerAndWait( makeResultFrame({ @@ -12764,7 +13147,22 @@ describe("ClaudeAdapterV2 background wake turns", () => { // workflow was re-attributed to) stops following it, so it is stopped // instead of left running out of sight. const settled = terminalReason === "api_error" ? "failed" : "cancelled"; - assert.deepEqual(stoppedTasks, [WORKFLOW_TASK_ID]); + assert.deepEqual( + stoppedTasks, + pendingStop ? [WORKFLOW_TASK_ID, WORKFLOW_TASK_ID] : [WORKFLOW_TASK_ID], + ); + if (stopping !== undefined) { + // Parent failure assigns a local outcome without a task_notification. + yield* TestClock.adjust("2 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(stopping))); + // Local parent cleanup still cannot provide a confirmed outcome on retry. + const retry = yield* harness.runtime.stopSubagent!({ + providerThread: harness.providerThread, + nativeTaskId: WORKFLOW_TASK_ID, + }).pipe(Effect.exit, Effect.forkScoped); + yield* TestClock.adjust("2 seconds"); + assert.isTrue(Exit.isFailure(yield* Fiber.join(retry))); + } assert.isFalse(yield* harness.hasPendingBackgroundWork); assert.equal(workflowCoordinatorEvents(harness.events).at(-1)?.subagent.status, settled); for (const index of [1, 2]) { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 7199f73040ef9fe8a5a2bd02efb963355ed4011d..880652b95f10ca824ebac422455d8b7776b4b932 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2957,6 +2957,8 @@ interface ActiveClaudeProviderRetry { interface ActiveClaudeSubagent { task: OrchestrationV2Subagent; + // Local parent-turn cleanup can end the card without a native task outcome. + readonly confirmedOutcome: ProviderAdapter.ProviderAdapterV2SubagentOutcome | undefined; // The launch run's root node. task.parentNodeId is that same node, or the // owning subagent's node for a subagent another subagent started. readonly rootNodeId: OrchestrationV2ExecutionNode["id"]; @@ -3497,6 +3499,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // idle, a running subagent's frames are projected through it at once // instead of waiting in the wake buffer for a continuation run. const settledTurnByNativeThread = new Map(); + const subagentStopOutcomes = new Map< + string, + Deferred.Deferred + >(); + // An idle outcome stays buffered for the parent's wake, but must not + // settle the child again after that child resumes. + const appliedSubagentOutcomes = new WeakSet(); // Subagent frames can precede the task_started that registers their // subagent (the same race rememberPendingClaudeSubagentLaunch covers). // They wait here and replay once task_started registers the owner. @@ -4419,6 +4428,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const ids = claudeSubagentIds(resume.context, resume.taskId); const now = yield* DateTime.now; const subagent: ActiveClaudeSubagent = { + confirmedOutcome: undefined, childThread: null, task: { id: ids.nodeId, @@ -4805,6 +4815,39 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } }); + const confirmedSubagentOutcome = (task: OrchestrationV2Subagent | undefined) => { + if ( + task === undefined || + task.completedAt === null || + (task.status !== "completed" && task.status !== "failed" && task.status !== "cancelled") + ) + return; + return { status: task.status, completedAt: task.completedAt, result: task.result }; + }; + + const discardBufferedSubagentStarts = ( + nativeThreadId: string, + taskId: string, + outcome: SDKMessage, + ) => + Ref.update(wakeBuffers, (current) => { + const entry = current.get(nativeThreadId); + if (entry === undefined) return current; + const outcomeIndex = entry.messages.indexOf(outcome); + const messages = entry.messages.filter( + (buffered, index) => + !( + (outcomeIndex === -1 || index < outcomeIndex) && + buffered.type === "system" && + (buffered.subtype === "task_started" || buffered.subtype === "task_progress") && + buffered.task_id === taskId + ), + ); + return messages.length === entry.messages.length + ? current + : new Map(current).set(nativeThreadId, { ...entry, messages }); + }); + const updateClaudeSubagentNode = Effect.fnUntraced(function* (input: { readonly context: ActiveClaudeTurnContext; readonly taskId: string; @@ -4827,6 +4870,10 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( "running" | "completed" | "failed" | "cancelled" >; readonly reopen?: boolean; + readonly nativeOutcome?: Extract< + SDKMessage, + { readonly type: "system"; readonly subtype: "task_notification" } + >; readonly onlyIfRunningInRun?: RunId | null; }) { // The session registry lets a wake-replay turn (fresh context maps) @@ -4993,6 +5040,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } satisfies OrchestrationV2Subagent; const subagent = { task, + confirmedOutcome: + input.nativeOutcome !== undefined + ? confirmedSubagentOutcome(task) + : isReopen + ? undefined + : existingSubagent?.confirmedOutcome, rootNodeId: existingSubagent?.rootNodeId ?? input.context.input.rootNodeId, childThreadId, childRootNodeId, @@ -5037,6 +5090,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return [true, new Map(current).set(input.taskId, subagent)] as const; }); if (!accepted) return; + if (input.nativeOutcome !== undefined) appliedSubagentOutcomes.add(input.nativeOutcome); // A call the subagent left open never returns once it ends. It ends // first, while the run that owns the child thread still takes its // events. @@ -5210,6 +5264,11 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }, }); + if (input.nativeOutcome !== undefined && subagent.confirmedOutcome !== undefined) { + const outcome = subagentStopOutcomes.get(input.taskId); + if (outcome !== undefined) yield* Deferred.succeed(outcome, subagent.confirmedOutcome); + } + // A completed subagent's result is normally its final assistant // message, which is already in the child thread when its text was // routed there. Failures and cancellations always get the message. @@ -6230,6 +6289,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const { progress: _staleProgress, ...priorTask } = registered.task; return new Map(current).set(message.task_id, { ...registered, + confirmedOutcome: undefined, task: { ...priorTask, status: "running", @@ -6699,6 +6759,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message, activeContext: null, }); + // Wake delivery must not delay a known child's actual outcome. + // Record wake evidence first, while the child is still running. + const registered = (yield* Ref.get(sessionSubagentsByTaskId)).get(message.task_id); + const settled = settledTurnByNativeThread.get(liveQuery.nativeThreadId); + const hasBufferedStart = + (yield* Ref.get(wakeBuffers)) + .get(liveQuery.nativeThreadId) + ?.messages.some( + (entry) => + entry.type === "system" && + entry.subtype === "task_started" && + entry.task_id === message.task_id, + ) ?? false; + // Starts and workflow transcripts still need the continuation's + // context. A pending Stop must receive its native outcome now. + const canApplyOutcome = + subagentStopOutcomes.has(message.task_id) || + (!hasBufferedStart && registered?.task.workflow === undefined); + if (registered !== undefined && settled !== undefined && canApplyOutcome) { + yield* discardBufferedSubagentStarts( + liveQuery.nativeThreadId, + message.task_id, + message, + ); + yield* updateClaudeSubagentNode({ + context: settled, + taskId: message.task_id, + ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), + workflowFrame: message, + nativeOutcome: message, + result: message.summary, + status: claudeTaskOutcome(message.status), + onlyIfRunningInRun: registered.task.runId, + }); + } } else { yield* applyBackgroundTaskRosterMessage({ nativeThreadId: liveQuery.nativeThreadId, @@ -7186,6 +7281,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } if (message.type === "system" && message.subtype === "task_notification") { + // A live outcome supersedes earlier idle starts. During replay only + // remove frames before this notification, preserving a later resume. + if ( + !appliedSubagentOutcomes.has(message) && + (yield* Ref.get(sessionSubagentsByTaskId)).has(message.task_id) + ) { + yield* discardBufferedSubagentStarts( + liveQuery.nativeThreadId, + message.task_id, + message, + ); + } // A wake-replay turn has empty ignoredTaskIds, so opaque-task // tracking (live roster, wake eligibility, or the short-lived // post-buffer replay tombstone) classifies local_bash before any @@ -7228,13 +7335,18 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( message, activeContext: context, }); - if (!wasBackgroundTask && !context.ignoredTaskIds.has(message.task_id)) { + if ( + !wasBackgroundTask && + !context.ignoredTaskIds.has(message.task_id) && + !appliedSubagentOutcomes.has(message) + ) { yield* updateClaudeSubagentNode({ context, taskId: message.task_id, ...(message.tool_use_id === undefined ? {} : { toolUseId: message.tool_use_id }), // Terminal usage totals arrive here, after the last snapshot. workflowFrame: message, + nativeOutcome: message, result: message.summary, status: message.status === "completed" @@ -8973,11 +9085,24 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const existing = yield* Ref.get(queryContext); const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId; const subagent = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); - if (subagent === undefined || subagent.task.status !== "running") return; + if (subagent?.task.threadId !== input.providerThread.appThreadId) { + return yield* new ProviderAdapter.ProviderAdapterProtocolError({ + driver: CLAUDE_PROVIDER, + detail: "Claude subagent has no known owning task to confirm its outcome.", + }); + } + if ( + input.requestedAt !== undefined && + subagent.task.startedAt !== null && + DateTime.toEpochMillis(subagent.task.startedAt) > + DateTime.toEpochMillis(input.requestedAt) + ) + return; + const confirmed = subagent.confirmedOutcome; + if (confirmed !== undefined) return confirmed; if ( existing === null || existing.nativeThreadId !== nativeThreadId || - subagent.task.threadId !== input.providerThread.appThreadId || existing.subagentsFromEarlierProcesses.has(subagent) || existing.query.stopTask === undefined ) { @@ -8986,16 +9111,41 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( detail: "Claude subagent has no live owning query with task stopping support.", }); } - yield* existing.query.stopTask(input.nativeTaskId).pipe( + const outcome = + subagentStopOutcomes.get(input.nativeTaskId) ?? + (yield* Deferred.make()); + subagentStopOutcomes.set(input.nativeTaskId, outcome); + const recorded = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); + if ( + recorded?.task.startedAt === subagent.task.startedAt && + recorded.confirmedOutcome !== undefined + ) { + if (subagentStopOutcomes.get(input.nativeTaskId) === outcome) + subagentStopOutcomes.delete(input.nativeTaskId); + return recorded.confirmedOutcome; + } + const confirmedOutcome = yield* existing.query.stopTask(input.nativeTaskId).pipe( + Effect.andThen(Deferred.await(outcome)), + Effect.timeout("2 seconds"), Effect.mapError( (cause) => new ProviderAdapter.ProviderAdapterProtocolError({ driver: CLAUDE_PROVIDER, - detail: "Claude native subagent stop failed.", + detail: "Claude subagent Stop was not confirmed by a native outcome.", cause, }), ), + Effect.ensuring( + Effect.sync(() => { + if (subagentStopOutcomes.get(input.nativeTaskId) === outcome) + subagentStopOutcomes.delete(input.nativeTaskId); + }), + ), ); + const current = (yield* Ref.get(sessionSubagentsByTaskId)).get(input.nativeTaskId); + return current?.task.startedAt === subagent.task.startedAt + ? confirmedOutcome + : undefined; }), respondToRuntimeRequest: Effect.fn("ClaudeAdapterV2.respondToRuntimeRequest")( function* (requestInput) { diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 230f4d4b77751bca240e5e81a6b4ed1ff36570b1..93cc4f6b6930c15767f3ab21ec0bee8d7a25be25 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -839,31 +839,71 @@ it.effect("settles a delegated child once its restart continuation fails for goo }), ); -it.effect("a native subagent stop does not settle its owner's background work", () => - Effect.gen(function* () { - const now = yield* DateTime.now; - const events = yield* Ref.make>([]); - const layer = layerExecutorFor({ - events, - interrupt: (input) => { - assert.equal(input.subagent?.nativeTaskId, "native-task"); - return Ref.update(events, (current) => [...current, "stop-child"]); - }, - threads: { dispatch: () => Effect.die("The owner's background work must stay running") }, - }); - yield* Effect.gen(function* () { - const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; - yield* executor.execute({ - ...restartEffect(now, { type: "detach" }), - request: { - type: "provider-turn.interrupt", - providerSessionId: oldSessionId, - providerThreadId, - providerTurnId, - subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" }, +it.effect.each(["confirmed", "unconfirmed", "failed"] as const)( + "a native subagent stop only repairs %s provider evidence", + (outcome) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const events = yield* Ref.make>([]); + const layer = layerExecutorFor({ + events, + interrupt: (input) => { + assert.equal(input.subagent?.nativeTaskId, "native-task"); + return Ref.update(events, (current) => [...current, "stop-child"]).pipe( + Effect.andThen( + outcome === "failed" + ? new ProviderTurnControlService.ProviderTurnControlError({ + threadId, + providerTurnId, + operation: "interrupt", + cause: "SDK stop failed", + }) + : Effect.succeed( + outcome === "confirmed" + ? { + status: "cancelled" as const, + completedAt: now, + result: "Stopped by Claude", + } + : undefined, + ), + ), + ); + }, + threads: { + dispatch: (command) => { + assert.equal(command.type, "thread.background-work.settle"); + if (command.type === "thread.background-work.settle") { + assert.equal(command.subagent?.id, "subagent"); + assert.deepEqual(command.subagent?.outcome, { + status: "cancelled", + completedAt: now, + result: "Stopped by Claude", + }); + } + return Ref.update(events, (current) => [...current, "settle-child"]).pipe( + Effect.as({ sequence: 1, storedEvents: [] }), + ); + }, }, }); - }).pipe(Effect.provide(layer)); - assert.deepEqual(yield* Ref.get(events), ["stop-child"]); - }), + const execution = yield* Effect.gen(function* () { + const executor = yield* EffectWorker.OrchestrationEffectExecutorV2; + yield* executor.execute({ + ...restartEffect(now, { type: "detach" }), + request: { + type: "provider-turn.interrupt", + providerSessionId: oldSessionId, + providerThreadId, + providerTurnId, + subagent: { id: NodeId.make("subagent"), nativeTaskId: "native-task" }, + }, + }); + }).pipe(Effect.provide(layer), Effect.exit); + assert.equal(Exit.isFailure(execution), outcome === "failed"); + assert.deepEqual( + yield* Ref.get(events), + outcome === "confirmed" ? ["stop-child", "settle-child"] : ["stop-child"], + ); + }), ); diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index 06ba83ca0d427bbf0f688c390331cf48c30cf8c6..aaf4507ca16853391fa63b771b6a9663e5d1a0d8 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -162,51 +162,73 @@ export const layerExecutor: Layer.Layer< }), ), ); - case "provider-turn.interrupt": + case "provider-turn.interrupt": { + const request = effect.request; return providerTurnControl .interrupt({ threadId: effect.threadId, - providerSessionId: effect.request.providerSessionId, - providerThreadId: effect.request.providerThreadId, - providerTurnId: effect.request.providerTurnId, - ...(effect.request.subagent === undefined + providerSessionId: request.providerSessionId, + providerThreadId: request.providerThreadId, + providerTurnId: request.providerTurnId, + ...(request.subagent === undefined ? {} - : { subagent: effect.request.subagent }), + : { + subagent: { + ...request.subagent, + requestedAt: DateTime.makeUnsafe(effect.createdAt), + }, + }), }) .pipe( + Effect.tapError((cause) => + request.subagent === undefined + ? Effect.void + : Effect.logWarning("orchestration-v2.subagent-stop-failed", { + effectId: effect.id, + cause, + }), + ), Effect.catch((cause) => + request.subagent === undefined && isNonRetryableProviderTurnControlFailure( - effect.request.type, + request.type, Cause.pretty(Cause.fail(cause)), ) ? Effect.void : Effect.fail(cause), ), - // The provider has stopped what it still ran and reported it. - // Whatever the thread still shows on that provider thread is - // work no process will report on, so the Stop ends it too. - // One Stop can interrupt several provider threads, so the - // settle is keyed by effect, not by the Stop command. - Effect.andThen( - effect.request.subagent !== undefined + // A targeted Stop repairs only an outcome the provider actually observed. + // A full-turn Stop retains its existing process-teardown repair. + Effect.andThen((outcome) => + request.subagent !== undefined && outcome === undefined ? Effect.void : threads.dispatch({ type: "thread.background-work.settle", commandId: CommandId.make(`${effect.id}:background-work-settled`), + ...(request.subagent === undefined || outcome === undefined + ? {} + : { + subagent: { + id: request.subagent.id, + stoppedAt: DateTime.makeUnsafe(effect.createdAt), + outcome, + }, + }), threadId: effect.threadId, - providerThreadId: effect.request.providerThreadId, - providerTurnId: effect.request.providerTurnId, + providerThreadId: request.providerThreadId, + providerTurnId: request.providerTurnId, }), ), Effect.mapError( (cause) => new OrchestrationEffectExecutionError({ effectId: effect.id, - effectType: effect.request.type, + effectType: request.type, cause, }), ), ); + } case "provider-turn.steer": return providerTurnControl .steer({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index d85504bb264e7e502fbdb088f4942e9e107a93e5..e1497473f1de94a8b993c543d69e99e8be570370 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -8917,6 +8917,60 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), ), ); + if (command.subagent !== undefined) { + const outcome = command.subagent.outcome; + const subagent = projection.subagents.find((task) => task.id === command.subagent?.id); + if ( + subagent?.origin !== "provider_native" || + !["pending", "running", "waiting"].includes(subagent.status) || + (subagent.startedAt !== null && + DateTime.toEpochMillis(subagent.startedAt) > + DateTime.toEpochMillis(command.subagent.stoppedAt)) + ) + return; + const now = yield* DateTime.now; + const allocateEventId = () => + idAllocator.allocate.event({ + threadId: command.threadId, + commandId: command.commandId, + }); + const ended = yield* endOrphanedNativeSubagent({ + projection, + subagentId: subagent.id, + result: outcome.result, + status: outcome.status, + completedAt: outcome.completedAt, + now, + emitted: yield* Ref.get(events), + allocateEventId, + }).pipe(mapDispatchError(command)); + yield* Ref.update(events, (current) => [...current, ...ended]); + if (subagent.childThreadId !== null) { + const child = yield* projectionStore + .getThreadRecords(subagent.childThreadId, ["nodes", "turnItems"], { + turnItemRunIds: [null], + turnItemStatuses: ["pending", "running", "waiting"], + }) + .pipe( + Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }), + mapDispatchError(command), + ); + if (child !== null) { + const childEnded = yield* endRunlessRootTurns({ + threadId: subagent.childThreadId, + providerInstanceId: subagent.providerInstanceId, + projection: child, + status: outcome.status, + completedAt: outcome.completedAt, + now, + emitted: yield* Ref.get(events), + allocateEventId, + }).pipe(mapDispatchError(command)); + yield* Ref.update(events, (current) => [...current, ...childEnded]); + } + } + return; + } const stoppedRunId = projection.attempts.find( (attempt) => attempt.id === stopped.providerTurn?.runAttemptId, )?.runId; @@ -9016,7 +9070,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const providerSessionId = providerThread.providerSessionId; const nativeTaskId = subagent.nativeTaskRef.nativeId; const now = yield* DateTime.now; - // Record the request with its effect; the provider notification supplies the outcome. + // The effect can repair stale projections only after the provider confirms an outcome. yield* emit( events, command, diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index eaf1a9f940d0d0a9774e315392a538bca7925961..9cc21037753ca7a3ef3a942bd133ad14cb079c7e 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -9,12 +9,14 @@ import { ThreadId, } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; +import type * as DateTime from "effect/DateTime"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; @@ -44,12 +46,19 @@ const isProviderTurnControlError = Schema.is(ProviderTurnControlError); export interface ProviderTurnControlServiceV2Shape { readonly interrupt: (input: { - readonly subagent?: { readonly id: NodeId; readonly nativeTaskId: string }; + readonly subagent?: { + readonly id: NodeId; + readonly nativeTaskId: string; + readonly requestedAt?: DateTime.Utc; + }; readonly threadId: ThreadId; readonly providerSessionId: ProviderSessionId; readonly providerThreadId: ProviderThreadId; readonly providerTurnId: ProviderTurnId; - }) => Effect.Effect; + }) => Effect.Effect< + void | ProviderAdapter.ProviderAdapterV2SubagentOutcome, + ProviderTurnControlError + >; readonly steer: (input: { readonly threadId: ThreadId; readonly providerSessionId: ProviderSessionId; @@ -178,7 +187,17 @@ export const layer: Layer.Layer< const session = Option.isSome(loaded.session) ? loaded.session : yield* sessions.get(input.providerSessionId); - if (Option.isNone(session)) return; + if (Option.isNone(session)) { + if (input.subagent !== undefined) { + return yield* new ProviderTurnControlError({ + threadId: input.threadId, + operation: "interrupt", + providerTurnId: input.providerTurnId, + cause: "No live provider session can confirm this subagent's outcome.", + }); + } + return; + } if (input.subagent !== undefined) { if (session.value.stopSubagent === undefined) { return yield* new ProviderTurnControlError({ @@ -188,11 +207,13 @@ export const layer: Layer.Layer< cause: "The provider does not support stopping a native subagent.", }); } - yield* session.value.stopSubagent({ + return yield* session.value.stopSubagent({ providerThread: loaded.providerThread, nativeTaskId: input.subagent.nativeTaskId, + ...(input.subagent.requestedAt === undefined + ? {} + : { requestedAt: input.subagent.requestedAt }), }); - return; } // A settled turn reaches its adapter too: only the adapter knows // whether it still runs work for the thread, and each one either diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index c0732ba2a54cf6f70533066bd42a15649afda3e7..73fc29a05569fda320df7790ab7f1e0dc4cda408 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -18,6 +18,8 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Stream from "effect/Stream"; +import * as EventStore from "./EventStore.ts"; import * as TestClock from "effect/testing/TestClock"; import * as SqlitePersistence from "../persistence/Sqlite.ts"; @@ -27,6 +29,8 @@ import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "./CommandReceiptStore.ts"; import { OrchestrationEffectRequestV2 } from "./EffectOutbox.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import * as EffectWorker from "./EffectWorker.ts"; +import * as EffectOutbox from "./EffectOutbox.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ThreadManagementService from "./ThreadManagementService.ts"; @@ -49,6 +53,8 @@ const layerTest = ThreadManagementService.layer.pipe( Layer.mergeAll( layerDatabase, ProjectionStore.layer.pipe(Layer.provide(layerDatabase)), + EventStore.layer.pipe(Layer.provide(layerDatabase)), + EffectOutbox.layer.pipe(Layer.provide(layerDatabase)), ProviderReplayHarness.layerWithRegistry( { name: "thread-stop" }, ProviderAdapterRegistry.layerFromAdapters([adapter]), @@ -589,9 +595,9 @@ it.effect("thread.stop marks a turn it cannot interrupt so a late agent watch is }).pipe(Effect.provide(layerTest)), ); -it.effect( - "a native child stop persists only its targeted effect and leaves its owner's queue and watch alone", - () => +it.effect.each(["missing session", "completed", "failed", "cancelled"] as const)( + "a native child Stop only settles a confirmed %s outcome and preserves its owner and sibling", + (providerState) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const projections = yield* ProjectionStore.ProjectionStoreV2; @@ -625,6 +631,49 @@ it.effect( occurredAt: now, payload: subagent, }); + const sibling = { + ...subagent, + id: NodeId.make("node:native-stop-sibling"), + childThreadId: null, + }; + yield* projections.apply({ + id: EventId.make("event:native-child-stop:sibling"), + type: "subagent.updated", + threadId, + runId: run.id, + nodeId: sibling.id, + occurredAt: now, + payload: sibling, + }); + const item = before.turnItems.find( + (item) => item.type === "subagent" && item.subagentId === subagent.id, + )!; + yield* projections.apply({ + id: EventId.make("event:native-child-stop:item"), + type: "turn-item.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: { ...item, status: "running", completedAt: null }, + }); + const childRoot = { + ...before.nodes.find((node) => node.id === run.rootNodeId)!, + id: NodeId.make("node:native-stop-child-root"), + threadId: subagent.childThreadId!, + runId: null, + parentNodeId: null, + status: "running" as const, + countsForRun: false, + completedAt: null, + }; + yield* projections.apply({ + id: EventId.make("event:native-child-stop:child-root"), + type: "node.updated", + threadId: childRoot.threadId, + nodeId: childRoot.id, + occurredAt: now, + payload: childRoot, + }); const providerThread = before.providerThreads.find( (thread) => thread.id === run.providerThreadId, )!; @@ -669,6 +718,16 @@ it.effect( }); yield* send(threadId, "queued", "queue_after_active"); const stateBefore = yield* threadState(threadId); + const eventStore = yield* EventStore.EventStoreV2; + const afterSequence = yield* eventStore.latestSequence({ threadId }); + const confirmedStatus = + providerState === "completed" || providerState === "failed" || providerState === "cancelled" + ? providerState + : null; + const confirmedOutcome = + confirmedStatus === null + ? undefined + : { status: confirmedStatus, completedAt: now, result: "Actual provider outcome" }; const commandId = CommandId.make("stop-native-child"); yield* orchestrator.dispatch({ type: "subagent.stop", @@ -703,5 +762,115 @@ it.effect( }), ); assert.isTrue(Exit.isFailure(unrelated)); + // The provider outcome predates the delayed projection repair. + yield* TestClock.adjust("1 second"); + // Only execute Stop's durable effect, with no terminal event ingestion. + yield* sql`DELETE FROM orchestration_v2_effect_outbox WHERE command_id != ${commandId}`; + if (providerState === "missing session") { + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + yield* worker.drain(); + } else { + yield* orchestrator.dispatch({ + type: "thread.background-work.settle", + commandId: CommandId.make("repair-native-child"), + threadId, + providerThreadId: providerThread.id, + providerTurnId, + subagent: { id: subagent.id, stoppedAt: now, outcome: confirmedOutcome! }, + }); + } + const settled = yield* orchestrator.getThreadProjection(threadId); + const cancelled = settled.subagents.find((task) => task.id === subagent.id)!; + if (confirmedOutcome === undefined) { + const events = yield* eventStore.read({ threadId, afterSequence }).pipe(Stream.runCollect); + assert.isFalse( + events.some( + ({ event }) => + (event.type === "subagent.updated" || + event.type === "node.updated" || + event.type === "turn-item.updated") && + ["completed", "failed", "cancelled", "interrupted"].includes(event.payload.status), + ), + ); + assert.equal(cancelled.status, "running"); + assert.equal( + (yield* orchestrator.getThreadProjection(childRoot.threadId)).nodes.find( + (node) => node.id === childRoot.id, + )?.status, + "running", + ); + assert.deepEqual(yield* threadState(threadId), stateBefore); + return; + } + + // Cached clients resume from this sequence, so all lifecycle repairs + // must be stored in the parent stream, not just applied to projections. + const repaired = yield* eventStore.read({ threadId, afterSequence }).pipe(Stream.runCollect); + const terminal = repaired.flatMap(({ event }) => + (event.type === "subagent.updated" || + event.type === "node.updated" || + event.type === "turn-item.updated") && + event.payload.status === confirmedStatus + ? [event] + : [], + ); + assert.deepEqual(terminal.map((event) => event.type).sort(), [ + "node.updated", + "subagent.updated", + "turn-item.updated", + ]); + for (const event of terminal) { + assert.equal(event.nodeId, subagent.id); + assert.equal( + DateTime.toEpochMillis(event.payload.completedAt!), + DateTime.toEpochMillis(confirmedOutcome!.completedAt), + ); + if (event.type === "subagent.updated" || event.type === "turn-item.updated") { + assert.isDefined(event.payload.updatedAt); + } + } + + assert.equal(cancelled.status, confirmedStatus); + assert.isNotNull(cancelled.completedAt); + assert.equal( + DateTime.toEpochMillis(cancelled.completedAt!), + DateTime.toEpochMillis(confirmedOutcome!.completedAt), + ); + assert.equal(cancelled.result, confirmedOutcome!.result); + assert.equal( + settled.turnItems.find((candidate) => candidate.id === item.id)?.status, + confirmedStatus, + ); + const child = yield* orchestrator.getThreadProjection(childRoot.threadId); + assert.equal(child.nodes.find((node) => node.id === childRoot.id)?.status, confirmedStatus); + assert.isNotNull(child.nodes.find((node) => node.id === childRoot.id)?.completedAt); + assert.equal(settled.subagents.find((task) => task.id === sibling.id)?.status, "running"); + assert.deepEqual(yield* threadState(threadId), stateBefore); + // An old Stop follow-up cannot cancel a genuine subsequent resume. + yield* TestClock.adjust("1 second"); + const resumedAt = yield* DateTime.now; + yield* projections.apply({ + id: EventId.make("event:native-child-stop:resume"), + type: "subagent.updated", + threadId, + runId: run.id, + nodeId: subagent.id, + occurredAt: resumedAt, + payload: { ...cancelled, status: "running", startedAt: resumedAt, completedAt: null }, + }); + yield* orchestrator.dispatch({ + type: "thread.background-work.settle", + commandId: CommandId.make("old-stop-after-resume"), + threadId, + providerThreadId: providerThread.id, + providerTurnId, + subagent: { id: subagent.id, stoppedAt: now, outcome: confirmedOutcome }, + }); + assert.equal( + (yield* orchestrator.getThreadProjection(threadId)).subagents.find( + (task) => task.id === subagent.id, + )?.status, + "running", + ); }).pipe(Effect.provide(layerTest)), ); diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 4822493765a8844905ba3baab8e9c0ee0e22898e..b4db1038f4ebf0761304f60603edfdf023eb2040 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3193,7 +3193,9 @@ const OrchestrationV2InternalCommand = Schema.Union([ * Follows a Stop once its provider returned: background work the settled * thread still shows on that provider thread is no longer reported by any * provider process, so it is marked interrupted. Only the stopped turn's run - * and older runs are settled; a later run's work is its own. + * and older runs are settled; a later run's work is its own. A subagent target + * limits repair to that task and its runless child work, using a confirmed + * provider outcome. */ Schema.Struct({ type: Schema.Literal("thread.background-work.settle"), @@ -3201,6 +3203,17 @@ const OrchestrationV2InternalCommand = Schema.Union([ threadId: ThreadId, providerThreadId: ProviderThreadId, providerTurnId: ProviderTurnId, + subagent: Schema.optional( + Schema.Struct({ + id: NodeId, + stoppedAt: Schema.DateTimeUtc, + outcome: Schema.Struct({ + status: Schema.Literals(["completed", "failed", "cancelled"]), + completedAt: Schema.DateTimeUtc, + result: Schema.NullOr(Schema.String), + }), + }), + ), }), /** * Stop for one thread, whatever it is doing: interrupts its running turn, holds its queue, diff --git a/packages/provider-core/src/server/ProviderAdapter.ts b/packages/provider-core/src/server/ProviderAdapter.ts index 809dbfac0110f33cee739d9e896e55e26e153052..7c316ff6c2847a0ab652c120655ebc0757648ef5 100644 --- a/packages/provider-core/src/server/ProviderAdapter.ts +++ b/packages/provider-core/src/server/ProviderAdapter.ts @@ -531,6 +531,13 @@ export interface ProviderAdapterV2McpApps { }) => Effect.Effect<{ readonly contents: ReadonlyArray }, ProviderAdapterV2Error>; } +/** A terminal task outcome observed by the provider, not inferred from a stop timeout. */ +export type ProviderAdapterV2SubagentOutcome = { + readonly status: "completed" | "failed" | "cancelled"; + readonly completedAt: NonNullable; + readonly result: OrchestrationV2Subagent["result"]; +}; + export interface ProviderAdapterV2SessionRuntime { readonly instanceId: ProviderInstanceId; readonly driver: ProviderDriverKind; @@ -606,7 +613,9 @@ export interface ProviderAdapterV2SessionRuntime { readonly stopSubagent?: (input: { readonly providerThread: OrchestrationV2ProviderThread; readonly nativeTaskId: string; - }) => Effect.Effect; + /** A delayed Stop must not interrupt a task resumed after this request. */ + readonly requestedAt?: NonNullable; + }) => Effect.Effect; /** * Lets a runtime shared by several app threads unload one provider thread's * native state (and its MCP servers) when that app thread detaches, while diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index 1a6b31a886e52cb40d6e2a4a5ea5b8acf2692165..a3a00c5005e2af1fbba9438106919dbacb8f1f3a 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -277,13 +277,15 @@ function writtenIds( export const endOrphanedNativeSubagent = (input: { readonly projection: Pick; readonly subagentId: NodeId; - readonly status: "interrupted" | "cancelled"; + readonly status: "completed" | "failed" | "interrupted" | "cancelled"; + readonly completedAt?: DateTime.Utc; + readonly result?: string | null; readonly now: DateTime.Utc; readonly emitted: ReadonlyArray; readonly allocateEventId: () => Effect.Effect; }) => Effect.gen(function* () { - const { now, status, subagentId } = input; + const { now, status, subagentId, completedAt = now } = input; const subagent = input.projection.subagents.find((entry) => entry.id === subagentId); if (subagent?.origin !== "provider_native") return []; const node = input.projection.nodes.find((entry) => entry.id === subagentId); @@ -307,7 +309,13 @@ export const endOrphanedNativeSubagent = (input: { id: yield* input.allocateEventId(), type: "subagent.updated", driver: subagent.driver, - payload: { ...subagent, status, completedAt: now, updatedAt: now }, + payload: { + ...subagent, + status, + completedAt, + updatedAt: now, + ...(input.result === undefined ? {} : { result: input.result }), + }, }); } if ( @@ -319,7 +327,7 @@ export const endOrphanedNativeSubagent = (input: { ...scope, id: yield* input.allocateEventId(), type: "node.updated", - payload: { ...node, status, completedAt: now }, + payload: { ...node, status, completedAt }, }); } if ( @@ -331,7 +339,13 @@ export const endOrphanedNativeSubagent = (input: { ...scope, id: yield* input.allocateEventId(), type: "turn-item.updated", - payload: { ...item, status, completedAt: now, updatedAt: now }, + payload: { + ...item, + status, + completedAt, + updatedAt: now, + ...(input.result === undefined ? {} : { result: input.result }), + }, }); } return events; @@ -346,13 +360,14 @@ export const endRunlessRootTurns = (input: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; readonly projection: Pick; - readonly status: "interrupted" | "cancelled"; + readonly status: "completed" | "failed" | "interrupted" | "cancelled"; + readonly completedAt?: DateTime.Utc; readonly now: DateTime.Utc; readonly emitted: ReadonlyArray; readonly allocateEventId: () => Effect.Effect; }) => Effect.gen(function* () { - const { now, status } = input; + const { now, status, completedAt = now } = input; const endedNodeIds = writtenIds(input.emitted, "node.updated"); const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); const scope = { @@ -377,7 +392,7 @@ export const endRunlessRootTurns = (input: { id: yield* input.allocateEventId(), type: "node.updated", nodeId: node.id, - payload: { ...node, status, completedAt: now }, + payload: { ...node, status, completedAt }, }); } for (const item of input.projection.turnItems) { @@ -396,8 +411,12 @@ export const endRunlessRootTurns = (input: { nodeId: node.id, payload: { ...item, - status, - completedAt: now, + status: + status === "completed" && + (item.type === "command_execution" || item.type === "dynamic_tool") + ? "failed" + : status, + completedAt, updatedAt: now, ...(item.type === "reasoning" || item.type === "assistant_message" ? { streaming: false } -- 2.54.0