From e70cfc0baca3faee7848cd7b4b94d39c7e1ab32f Mon Sep 17 00:00:00 2001 From: Rui Conti Date: Sat, 1 Aug 2026 15:25:26 -0400 Subject: [PATCH 1/2] feat(eve): run subagents as background tasks behind experimental.tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With the flag on, every subagent dispatch (fresh start, agentId continuation, local and remote) becomes delegated execution: - the durable task run is created before the child dispatch side effect, so a fast child always finds a live command hook; a replay re-derives the same deterministic command token and the duplicate run loses the hook claim; - the child's reply address is the task run's private hook instead of the parent turn's inbox, so the existing local notification, HITL forwarding, and remote callback wires deliver to the task unchanged; - the originating call resolves immediately with a task receipt whose parked outcome settles the agent handle through the existing resolve path, keeping history provider-valid with exactly one result; - a dispatch that never acknowledges a child fails its task out of band and returns the failure directly, so the model never sees a task id for work that never started. The task run translates the child wire into lifecycle commands (settled turn outcomes to complete/fail/cancel, HITL batches to input_required, authorization events to blocked/unblocked) and wakes the parent through the ordinary session delivery path when a task becomes ready — a parked parent starts a turn, an active turn observes the notification at its next safe boundary. task_peek and task_cancel execute inside the dispatch step, which owns the session task index and world access; task_cancel commits cancelled before propagating a cooperative abort through the handle address. task_await rides the turn's existing inbox wait: a small aggregation run polls the selected snapshot streams and posts the one tool-result the pending key expects, exiting quietly when the waiting turn is gone. Parent finalization cancels live tasks before terminating children. The flag also implies conversation-mode children so experimental.tasks and experimental.subagentPersistentSessions never produce a third mode. Child usage accounting for delegated tasks is deferred; the subagent-start machinery moved to its own module to keep the dispatch step within the file-length cap. Without the flag, dispatch behavior, wire shapes, and events are unchanged (agent-messaging scenario suite passes unmodified). Signed-off-by: Rui Conti --- .../dispatch-runtime-actions-step.ts | 83 ++++- .../dispatch-workflow-runtime-actions-step.ts | 4 +- packages/eve/src/execution/tasks/dispatch.ts | 329 ++++++++++++++++++ .../eve/src/execution/tasks/run-control.ts | 30 +- packages/eve/src/execution/tasks/run-steps.ts | 70 ++++ .../src/execution/tasks/run-workflow.test.ts | 69 +++- .../eve/src/execution/tasks/run-workflow.ts | 41 ++- .../terminate-child-sessions-step.ts | 32 ++ packages/eve/src/tasks/json.ts | 43 +++ packages/eve/src/tasks/task-id.ts | 22 ++ packages/eve/src/tasks/transitions.test.ts | 20 ++ packages/eve/src/tasks/transitions.ts | 13 + packages/eve/src/tasks/types.ts | 59 +++- packages/eve/src/tasks/wire.test.ts | 105 ++++++ packages/eve/src/tasks/wire.ts | 54 +++ 15 files changed, 939 insertions(+), 35 deletions(-) create mode 100644 packages/eve/src/execution/tasks/dispatch.ts create mode 100644 packages/eve/src/tasks/json.ts create mode 100644 packages/eve/src/tasks/wire.test.ts create mode 100644 packages/eve/src/tasks/wire.ts diff --git a/packages/eve/src/execution/dispatch-runtime-actions-step.ts b/packages/eve/src/execution/dispatch-runtime-actions-step.ts index 4fe974ada2..bd9e7cd5b3 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-step.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-step.ts @@ -47,11 +47,19 @@ import { } from "#protocol/message.js"; import type { RuntimeActionRequest, + RuntimeActionResult, RuntimeRemoteAgentCallActionRequest, RuntimeSubagentCallActionRequest, RuntimeSubagentDispatchFailure, - RuntimeSubagentResult, + RuntimeToolCallActionRequest, } from "#runtime/actions/types.js"; +import { + beginDelegatedTask, + executeTaskControlAction, + failDelegatedDispatch, + isTaskControlAction, + settleDelegatedDispatch, +} from "#execution/tasks/dispatch.js"; import { createDurableSessionState, type DurableSessionState, @@ -102,7 +110,8 @@ type DispatchPlanEntry = readonly dynamicRemoteAgent?: DynamicRemoteAgentConfig; } | { readonly kind: "reject"; readonly result: RuntimeSubagentDispatchFailure } - | { readonly kind: "start"; readonly target: DispatchStartTarget }; + | { readonly kind: "start"; readonly target: DispatchStartTarget } + | { readonly kind: "task-control"; readonly action: RuntimeToolCallActionRequest }; type DispatchStartTarget = | { @@ -125,7 +134,7 @@ export async function dispatchRuntimeActionsStep(input: { readonly serializedContext: Record; readonly sessionState: DurableSessionState; }): Promise<{ - readonly results: readonly RuntimeSubagentResult[]; + readonly results: readonly RuntimeActionResult[]; readonly sessionState: DurableSessionState; }> { "use step"; @@ -157,8 +166,12 @@ export async function dispatchRuntimeActionsStep(input: { // Read here, not in the child: trace state is scoped to one session's // context, so this is the last place the parent's window is visible. const parentTraceContext = readSessionTraceContext(input.serializedContext, session.sessionId); + const tasksEnabled = bundle.resolvedAgent.config.experimental?.tasks === true; + // Background tasks require resumable children: the flag implies + // conversation-mode dispatch so `experimental.tasks` and + // `experimental.subagentPersistentSessions` never produce a third mode. const persistentSessions = - bundle.resolvedAgent.config.experimental?.subagentPersistentSessions === true; + tasksEnabled || bundle.resolvedAgent.config.experimental?.subagentPersistentSessions === true; // A corrupt handle store throws; surface that before anything dispatches. // A mid-loop throw after a sibling started would durably replay the whole // batch and re-dispatch that sibling. @@ -177,7 +190,7 @@ export async function dispatchRuntimeActionsStep(input: { ).length; let nextSession = session; - const results: RuntimeSubagentResult[] = []; + const results: RuntimeActionResult[] = []; try { for (const entry of plan) { @@ -186,6 +199,31 @@ export async function dispatchRuntimeActionsStep(input: { continue; } + if (entry.kind === "task-control") { + const control = await executeTaskControlAction({ + action: entry.action, + bundle, + session: nextSession, + }); + if (control.result !== undefined) { + results.push(control.result); + } + continue; + } + + // Delegated execution: the durable task record exists before the + // child dispatch side effect, and the child's reply address is the + // task run's private hook instead of the parent turn's inbox. + const delegated = tasksEnabled + ? await beginDelegatedTask({ + ...describeDelegatedEntry(entry), + parentSessionId: session.sessionId, + parentTurnId: batch.event.turnId, + session: nextSession, + }) + : undefined; + const delegatedParentToken = delegated?.commandToken; + let outcome: DispatchOutcome; switch (entry.kind) { case "resume": @@ -198,7 +236,8 @@ export async function dispatchRuntimeActionsStep(input: { dynamicRemoteAgent: entry.dynamicRemoteAgent, }), currentSession: nextSession, - parentToken: input.parentContinuationToken ?? session.continuationToken, + parentToken: + delegatedParentToken ?? input.parentContinuationToken ?? session.continuationToken, parentTurnId: batch.event.turnId, }); break; @@ -213,7 +252,7 @@ export async function dispatchRuntimeActionsStep(input: { currentSession: nextSession, fanoutSize, initiatorAuth, - parentContinuationToken: input.parentContinuationToken, + parentContinuationToken: delegatedParentToken ?? input.parentContinuationToken, parentTraceContext, persistentSessions, session, @@ -224,10 +263,25 @@ export async function dispatchRuntimeActionsStep(input: { nextSession = outcome.session; if (outcome.kind === "error") { + if (delegated !== undefined) { + await failDelegatedDispatch({ error: outcome.result.output, task: delegated }); + } results.push(outcome.result); continue; } + if (delegated !== undefined) { + const settled = await settleDelegatedDispatch({ + callId: outcome.callId, + childSessionId: outcome.address.sessionId, + session: nextSession, + subagentName: outcome.toolName, + task: delegated, + }); + nextSession = settled.session; + results.push(settled.receipt); + } + // Emission is observability, not control flow: a failure here must not // escape the loop, because a durable-step retry would re-dispatch the // children that already started. @@ -295,6 +349,10 @@ function planDispatch(input: { const handles = getAgentHandleStore(input.session.state)?.handles ?? []; return input.actions.map((action): DispatchPlanEntry => { + if (isTaskControlAction(action)) { + return { action, kind: "task-control" }; + } + const rawAgentId = action.input.agentId; const agentId = typeof rawAgentId === "string" && rawAgentId.trim() !== "" ? rawAgentId : undefined; @@ -686,6 +744,17 @@ async function startRemoteSubagent(input: { } } +/** Names one delegated dispatch for its task record, before any child exists. */ +function describeDelegatedEntry(entry: Extract): { + readonly callId: string; + readonly mode: "local" | "remote"; + readonly name: string; +} { + const action = entry.kind === "resume" ? entry.action : entry.target.action; + return action.kind === "remote-agent-call" + ? { callId: action.callId, mode: "remote", name: action.remoteAgentName } + : { callId: action.callId, mode: "local", name: action.subagentName }; +} function isRecursiveAgentAction( action: RuntimeActionRequest, subagentsByNodeId: ReadonlyMap, diff --git a/packages/eve/src/execution/dispatch-workflow-runtime-actions-step.ts b/packages/eve/src/execution/dispatch-workflow-runtime-actions-step.ts index 47379b3f63..55ea3c62b7 100644 --- a/packages/eve/src/execution/dispatch-workflow-runtime-actions-step.ts +++ b/packages/eve/src/execution/dispatch-workflow-runtime-actions-step.ts @@ -19,7 +19,7 @@ import { BundleKey } from "#runtime/sessions/runtime-context-keys.js"; import type { RuntimeActionRequest, RuntimeSubagentDispatchFailure, - RuntimeSubagentResult, + RuntimeActionResult, } from "#runtime/actions/types.js"; import { resolveEffectiveAgentRuntime } from "#execution/effective-agent-config.js"; @@ -33,7 +33,7 @@ export async function dispatchWorkflowRuntimeActionsStep(input: { readonly serializedContext: Record; readonly sessionState: DurableSessionState; }): Promise<{ - readonly results: readonly RuntimeSubagentResult[]; + readonly results: readonly RuntimeActionResult[]; readonly sessionState: DurableSessionState; }> { "use step"; diff --git a/packages/eve/src/execution/tasks/dispatch.ts b/packages/eve/src/execution/tasks/dispatch.ts new file mode 100644 index 0000000000..f3a1858ea9 --- /dev/null +++ b/packages/eve/src/execution/tasks/dispatch.ts @@ -0,0 +1,329 @@ +import type { RuntimeSession } from "#execution/agent-handle-dispatch.js"; +import { + cancelRemoteAgentTurn, + resolveRemoteAgentForAction, +} from "#execution/remote-agent-dispatch.js"; +import { + readLatestTaskSnapshot, + sendTaskCommand, + startTaskRun, +} from "#execution/tasks/run-control.js"; +import { requestWorkflowTurnCancellation } from "#execution/workflow-runtime.js"; +import { getAgentHandleStore, type AgentHandle } from "#harness/handles/store.js"; +import { createLogger, logError } from "#internal/logging.js"; +import type { + RuntimeActionRequest, + RuntimeActionResult, + RuntimeSubagentChildResult, + RuntimeToolCallActionRequest, +} from "#runtime/actions/types.js"; +import type { CompiledBundle } from "#runtime/sessions/runtime-context-keys.js"; +import { + TASK_CANCEL_TOOL_NAME, + TASK_CONTROL_TOOL_NAMES, + TASK_PEEK_TOOL_NAME, +} from "#runtime/framework-tools/tasks.js"; +import type { JsonValue } from "#shared/json.js"; +import { taskViewsToJson } from "#tasks/json.js"; +import { + findSessionTaskEntry, + recordSessionTask, + type SessionTaskIndexEntry, +} from "#tasks/session-index.js"; +import { deriveTaskCommandToken, deriveTaskId } from "#tasks/task-id.js"; +import { isReadyTaskStatus, type TaskView } from "#tasks/types.js"; + +const log = createLogger("execution.tasks.dispatch"); + +const CANCEL_COMMIT_POLL_ATTEMPTS = 10; +const CANCEL_COMMIT_POLL_DELAY_MS = 250; + +/** A prepared delegated task: identity plus its started durable run. */ +export interface DelegatedTask { + readonly commandToken: string; + readonly taskId: string; + readonly taskRunId: string; +} + +/** True for `task_peek` / `task_cancel` calls. */ +export function isTaskControlAction( + action: RuntimeActionRequest, +): action is RuntimeToolCallActionRequest { + return action.kind === "tool-call" && TASK_CONTROL_TOOL_NAMES.has(action.toolName); +} + +/** + * Creates the durable task record for one delegated subagent call, + * before the child dispatch side effect. The task must exist first so + * a fast child always finds a live command hook; a duplicate replay + * re-derives the same token and the loser exits on the hook claim. + */ +export async function beginDelegatedTask(input: { + readonly callId: string; + readonly mode: "local" | "remote"; + readonly name: string; + readonly parentSessionId: string; + readonly parentTurnId: string; + readonly session: RuntimeSession; +}): Promise { + const taskId = deriveTaskId({ + callId: input.callId, + parentSessionId: input.parentSessionId, + parentTurnId: input.parentTurnId, + }); + const commandToken = deriveTaskCommandToken({ + parentContinuationToken: input.session.continuationToken, + taskId, + }); + const run = await startTaskRun({ + commandToken, + initialView: { + metadata: { kind: "subagent", mode: input.mode, name: input.name }, + status: "working", + taskId, + }, + wakeToken: input.session.continuationToken, + }); + return { commandToken, taskId, taskRunId: run.runId }; +} + +/** + * Settles a delegated dispatch that acknowledged a child: attaches the + * child session to the task, records the task in the session index, and + * returns the receipt that resolves the originating tool call. + * + * The receipt carries a `parked` outcome so the existing resolve path + * settles the agent handle to `parked` — the handle keeps the child + * address for follow-ups while the task run owns the outstanding work. + */ +export async function settleDelegatedDispatch(input: { + readonly callId: string; + readonly childSessionId: string; + readonly session: RuntimeSession; + readonly subagentName: string; + readonly task: DelegatedTask; +}): Promise<{ readonly receipt: RuntimeSubagentChildResult; readonly session: RuntimeSession }> { + // The freshly started task run may not have registered its hook yet; + // ride out that startup window instead of dropping the acknowledgement. + await sendTaskCommand({ + command: { childSessionId: input.childSessionId, kind: "describe" }, + commandToken: input.task.commandToken, + retryUnreachable: { attempts: 20, delayMs: 250 }, + }); + const receiptOutput = { status: "working", taskId: input.task.taskId }; + return { + receipt: { + callId: input.callId, + kind: "subagent-result", + origin: "child", + outcome: { + kind: "parked", + result: { + kind: "succeeded", + output: `Delegated as background task ${input.task.taskId} (working).`, + }, + usageDelta: { cacheReadTokens: 0, cacheWriteTokens: 0, inputTokens: 0, outputTokens: 0 }, + }, + output: receiptOutput, + subagentName: input.subagentName, + }, + session: recordSessionTask(input.session, { + commandToken: input.task.commandToken, + taskId: input.task.taskId, + taskRunId: input.task.taskRunId, + }), + }; +} + +/** + * Terminates the task record for a dispatch that never acknowledged a + * child. The originating call gets the dispatch failure directly; the + * task fails out of band and is never recorded in the session index, + * so the model never sees a task id for work that never started. + */ +export async function failDelegatedDispatch(input: { + readonly error: JsonValue; + readonly task: DelegatedTask; +}): Promise { + await sendTaskCommand({ + command: { data: input.error, kind: "fail" }, + commandToken: input.task.commandToken, + retryUnreachable: { attempts: 20, delayMs: 250 }, + }); +} + +/** + * Executes one task-control call inside the dispatch step, which holds + * the durable session state (ownership index) and world access the + * tools need. + */ +export async function executeTaskControlAction(input: { + readonly action: RuntimeToolCallActionRequest; + readonly bundle: CompiledBundle; + readonly session: RuntimeSession; +}): Promise<{ readonly result: RuntimeActionResult | undefined }> { + const { action } = input; + const taskIds = readTaskIds(action.input); + if (taskIds === undefined || taskIds.length === 0) { + return { + result: createTaskControlError(action, "Provide a non-empty `taskIds` array."), + }; + } + + const entries: SessionTaskIndexEntry[] = []; + const unknown: string[] = []; + for (const taskId of taskIds) { + const entry = findSessionTaskEntry(input.session.state, taskId); + if (entry === undefined) { + unknown.push(taskId); + } else { + entries.push(entry); + } + } + if (unknown.length > 0) { + return { + result: createTaskControlError( + action, + `Unknown task ids: ${unknown.join(", ")}. Tasks belong to the session that created them.`, + ), + }; + } + + switch (action.toolName) { + case TASK_PEEK_TOOL_NAME: { + const views = await readTaskViews(entries); + return { result: createTaskViewsResult(action, views) }; + } + case TASK_CANCEL_TOOL_NAME: { + const views = await Promise.all( + entries.map((entry) => + cancelOneTask({ bundle: input.bundle, entry, session: input.session }), + ), + ); + return { result: createTaskViewsResult(action, views) }; + } + default: + return { + result: createTaskControlError(action, `Unsupported task control "${action.toolName}".`), + }; + } +} + +async function cancelOneTask(input: { + readonly bundle: CompiledBundle; + readonly entry: SessionTaskIndexEntry; + readonly session: RuntimeSession; +}): Promise { + const { entry } = input; + await sendTaskCommand({ command: { kind: "cancel" }, commandToken: entry.commandToken }); + + // The `cancelled` state must commit before the executor abort + // propagates, so a late child result can never revive the task. + let view = await readLatestTaskSnapshot({ taskRunId: entry.taskRunId }); + for ( + let attempt = 0; + attempt < CANCEL_COMMIT_POLL_ATTEMPTS && + !(view !== undefined && isReadyTaskStatus(view.status)); + attempt += 1 + ) { + await new Promise((resolve) => setTimeout(resolve, CANCEL_COMMIT_POLL_DELAY_MS)); + view = await readLatestTaskSnapshot({ taskRunId: entry.taskRunId }); + } + const settledView = view ?? createPendingTaskView(entry.taskId); + + if (settledView.status === "cancelled") { + await propagateTaskCancel({ bundle: input.bundle, session: input.session, view: settledView }); + } + return settledView; +} + +/** + * Best-effort cooperative abort of the cancelled task's child turn, + * routed through the agent handle that owns the child address. A task + * whose handle is already gone has nothing left to abort. + */ +async function propagateTaskCancel(input: { + readonly bundle: CompiledBundle; + readonly session: RuntimeSession; + readonly view: TaskView; +}): Promise { + const childSessionId = input.view.metadata.childSessionId; + if (childSessionId === undefined) return; + const handles = getAgentHandleStore(input.session.state)?.handles ?? []; + const handle = handles + .filter( + (candidate): candidate is Extract => + candidate.phase === "running" || candidate.phase === "parked", + ) + .find((candidate) => candidate.address.sessionId === childSessionId); + + try { + if (handle !== undefined && handle.address.kind === "agent/remote") { + const resolved = resolveRemoteAgentForAction({ + nodeId: handle.identity.nodeId, + remoteAgentName: handle.identity.name, + registry: input.bundle.subagentRegistry.subagentsByNodeId, + }); + await cancelRemoteAgentTurn({ + remote: { ...resolved, url: handle.address.url }, + sessionId: childSessionId, + }); + return; + } + await requestWorkflowTurnCancellation({ sessionId: childSessionId }); + } catch (error) { + logError(log, "task cancel propagation failed; the child may run to completion", error, { + childSessionId, + taskId: input.view.taskId, + }); + } +} + +async function readTaskViews(entries: readonly SessionTaskIndexEntry[]): Promise { + return Promise.all( + entries.map( + async (entry) => + (await readLatestTaskSnapshot({ taskRunId: entry.taskRunId })) ?? + createPendingTaskView(entry.taskId), + ), + ); +} + +function createPendingTaskView(taskId: string): TaskView { + return { + metadata: { kind: "subagent", mode: "local", name: "unknown" }, + status: "working", + taskId, + }; +} + +function createTaskViewsResult( + action: RuntimeToolCallActionRequest, + views: readonly TaskView[], +): RuntimeActionResult { + return { + callId: action.callId, + kind: "tool-result", + output: taskViewsToJson(views), + toolName: action.toolName, + }; +} + +function createTaskControlError( + action: RuntimeToolCallActionRequest, + message: string, +): RuntimeActionResult { + return { + callId: action.callId, + isError: true, + kind: "tool-result", + output: { message }, + toolName: action.toolName, + }; +} + +function readTaskIds(input: Record): readonly string[] | undefined { + const value = input.taskIds; + if (!Array.isArray(value)) return undefined; + return value.filter((id): id is string => typeof id === "string" && id.trim() !== ""); +} diff --git a/packages/eve/src/execution/tasks/run-control.ts b/packages/eve/src/execution/tasks/run-control.ts index d3e00912c1..5e14ac1269 100644 --- a/packages/eve/src/execution/tasks/run-control.ts +++ b/packages/eve/src/execution/tasks/run-control.ts @@ -39,23 +39,33 @@ export async function startTaskRun( /** * Submits one command to a task run. * - * `unreachable` means the run already finished and disposed its hook — - * the task is terminal, and the caller should read the final snapshot - * instead of treating the send as a failure. + * `unreachable` means the hook is not resumable — either the run + * already finished and disposed it (the task is terminal; read the + * final snapshot) or, right after creation, the freshly started run has + * not registered it yet. Senders racing that startup window pass + * `retryUnreachable`; senders addressing an established task treat + * `unreachable` as the terminal signal. */ export async function sendTaskCommand(input: { readonly command: TaskCommand; readonly commandToken: string; + readonly retryUnreachable?: { readonly attempts: number; readonly delayMs: number }; }): Promise<"delivered" | "unreachable"> { const payload: TaskCommandHookPayload = { command: input.command, kind: "task-command" }; - try { - await resumeHook(input.commandToken, payload); - return "delivered"; - } catch (error) { - if (isFinishedTaskRunTarget(error)) { - return "unreachable"; + const attempts = Math.max(1, input.retryUnreachable?.attempts ?? 1); + for (let attempt = 0; ; attempt += 1) { + try { + await resumeHook(input.commandToken, payload); + return "delivered"; + } catch (error) { + if (!isFinishedTaskRunTarget(error)) { + throw error; + } + if (attempt + 1 >= attempts) { + return "unreachable"; + } + await new Promise((resolve) => setTimeout(resolve, input.retryUnreachable?.delayMs ?? 250)); } - throw error; } } diff --git a/packages/eve/src/execution/tasks/run-steps.ts b/packages/eve/src/execution/tasks/run-steps.ts index faa2ebbb6b..0c8c50af16 100644 --- a/packages/eve/src/execution/tasks/run-steps.ts +++ b/packages/eve/src/execution/tasks/run-steps.ts @@ -1,7 +1,19 @@ import { getWritable } from "#compiled/@workflow/core/index.js"; +import { + EntityConflictError, + HookNotFoundError, + RunExpiredError, + WorkflowRunNotFoundError, +} from "#compiled/@workflow/errors/index.js"; +import type { DeliverHookPayload } from "#channel/types.js"; +import { resumeHook } from "#internal/workflow/runtime.js"; +import { createLogger } from "#internal/logging.js"; +import { walkCauseChain } from "#shared/errors.js"; import { TASK_SNAPSHOT_STREAM_NAMESPACE, type TaskView } from "#tasks/types.js"; +const log = createLogger("execution.tasks.run"); + /** * Appends one full task snapshot to the owning task run's `eve.task` * stream. Only the task run workflow calls this, which is what makes @@ -18,3 +30,61 @@ export async function appendTaskSnapshotStep(input: { readonly view: TaskView }) writer.releaseLock(); } } + +/** + * Wakes the parent session with a framework task notification. + * + * Rides the ordinary session delivery path: a parked parent starts a + * turn carrying this message, while an active turn observes it at the + * next safe boundary through the driver's normal delivery routing. A + * parent whose session already ended is a tolerated no-op. + */ +export async function wakeTaskParentStep(input: { + readonly token: string; + readonly view: TaskView; +}): Promise { + "use step"; + + const payload: DeliverHookPayload = { + kind: "deliver", + payloads: [ + { + message: formatTaskNotification(input.view), + }, + ], + }; + try { + await resumeHook(input.token, payload); + } catch (error) { + if (isGoneParentTarget(error)) { + log.warn("task wake target is gone; the parent session already ended", { + status: input.view.status, + taskId: input.view.taskId, + }); + return; + } + throw error; + } +} + +function formatTaskNotification(view: TaskView): string { + const subject = `Background task ${view.taskId} (${view.metadata.name})`; + if (view.status === "input_required") { + return `${subject} needs input. Use task_peek to inspect the outstanding requests.`; + } + return `${subject} is ${view.status}. Use task_peek to read its output.`; +} + +function isGoneParentTarget(error: unknown): boolean { + for (const candidate of walkCauseChain(error)) { + if ( + HookNotFoundError.is(candidate) || + WorkflowRunNotFoundError.is(candidate) || + RunExpiredError.is(candidate) || + EntityConflictError.is(candidate) + ) { + return true; + } + } + return false; +} diff --git a/packages/eve/src/execution/tasks/run-workflow.test.ts b/packages/eve/src/execution/tasks/run-workflow.test.ts index f5725eef44..f540f071e3 100644 --- a/packages/eve/src/execution/tasks/run-workflow.test.ts +++ b/packages/eve/src/execution/tasks/run-workflow.test.ts @@ -2,9 +2,9 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { createHook, type Hook } from "#compiled/@workflow/core/index.js"; import { claimHookOwnership, disposeHook } from "#execution/hook-ownership.js"; -import { appendTaskSnapshotStep } from "#execution/tasks/run-steps.js"; +import { appendTaskSnapshotStep, wakeTaskParentStep } from "#execution/tasks/run-steps.js"; import { taskRunWorkflow } from "#execution/tasks/run-workflow.js"; -import type { TaskCommandHookPayload, TaskView } from "#tasks/types.js"; +import type { TaskCommandHookPayload, TaskRunInboundPayload, TaskView } from "#tasks/types.js"; vi.mock("#compiled/@workflow/core/index.js", () => ({ createHook: vi.fn(), @@ -18,6 +18,7 @@ vi.mock("../hook-ownership.js", async (importOriginal) => ({ vi.mock("./run-steps.js", () => ({ appendTaskSnapshotStep: vi.fn(), + wakeTaskParentStep: vi.fn(), })); afterEach(() => { @@ -37,7 +38,7 @@ function createWorkingView(): TaskView { }; } -function mockCommandHook(payloads: readonly TaskCommandHookPayload[]): void { +function mockCommandHook(payloads: readonly TaskRunInboundPayload[]): void { const queue = [...payloads]; const hook = { [Symbol.asyncIterator]: () => ({ @@ -47,7 +48,7 @@ function mockCommandHook(payloads: readonly TaskCommandHookPayload[]): void { : { done: true as const, value: undefined }, }), token: "task-token", - } as Hook; + } as Hook; vi.mocked(createHook).mockReturnValue(hook); } @@ -107,4 +108,64 @@ describe("taskRunWorkflow", () => { expect(appendedStatuses()).toEqual(["working", "input_required"]); expect(disposeHook).toHaveBeenCalledTimes(1); }); + + it("translates a settled child turn from the wire and wakes the parent once ready", async () => { + const ZERO = { cacheReadTokens: 0, cacheWriteTokens: 0, inputTokens: 0, outputTokens: 0 }; + mockCommandHook([ + { command: { childSessionId: "child-session-1", kind: "describe" }, kind: "task-command" }, + { + kind: "runtime-action-result", + results: [ + { + outcome: { + kind: "parked", + result: { kind: "succeeded", output: "answer" }, + usageDelta: ZERO, + }, + output: "answer", + }, + ], + }, + ]); + + await taskRunWorkflow({ + commandToken: "task-token", + initialView: { + ...createWorkingView(), + metadata: { kind: "subagent", mode: "local", name: "research" }, + }, + wakeToken: "parent-session-token", + }); + + expect(appendedStatuses()).toEqual(["working", "working", "completed"]); + expect(wakeTaskParentStep).toHaveBeenCalledTimes(1); + expect(vi.mocked(wakeTaskParentStep).mock.calls[0]?.[0]).toMatchObject({ + token: "parent-session-token", + view: { status: "completed", taskId: "task_abc123" }, + }); + }); + + it("does not wake without a wake token and never wakes twice for one blocked child", async () => { + mockCommandHook([ + { command: { inputRequests: [{ q: 1 }], kind: "require-input" }, kind: "task-command" }, + { command: { inputRequests: [{ q: 2 }], kind: "require-input" }, kind: "task-command" }, + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ + commandToken: "task-token", + initialView: createWorkingView(), + wakeToken: "parent-session-token", + }); + + // input_required wakes once; the second require-input replaces the + // batch without leaving the ready state, and completing from ready + // does not re-wake. + expect(wakeTaskParentStep).toHaveBeenCalledTimes(1); + + vi.mocked(wakeTaskParentStep).mockClear(); + mockCommandHook([{ command: { data: "done", kind: "complete" }, kind: "task-command" }]); + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + expect(wakeTaskParentStep).not.toHaveBeenCalled(); + }); }); diff --git a/packages/eve/src/execution/tasks/run-workflow.ts b/packages/eve/src/execution/tasks/run-workflow.ts index b8df99788f..4d9eaf6a47 100644 --- a/packages/eve/src/execution/tasks/run-workflow.ts +++ b/packages/eve/src/execution/tasks/run-workflow.ts @@ -1,9 +1,15 @@ import { createHook } from "#compiled/@workflow/core/index.js"; import { claimHookOwnership, disposeHook, isHookConflictError } from "#execution/hook-ownership.js"; -import { appendTaskSnapshotStep } from "#execution/tasks/run-steps.js"; +import { appendTaskSnapshotStep, wakeTaskParentStep } from "#execution/tasks/run-steps.js"; import { applyTaskTransition } from "#tasks/transitions.js"; -import { isTerminalTaskStatus, type TaskCommandHookPayload, type TaskView } from "#tasks/types.js"; +import { translateTaskInboundPayload } from "#tasks/wire.js"; +import { + isReadyTaskStatus, + isTerminalTaskStatus, + type TaskRunInboundPayload, + type TaskView, +} from "#tasks/types.js"; /** Input for one durable task run. */ export interface TaskRunWorkflowInput { @@ -11,16 +17,27 @@ export interface TaskRunWorkflowInput { readonly commandToken: string; /** The creation snapshot, normally `working`. */ readonly initialView: TaskView; + /** + * Parent session delivery token used to wake a parked parent when the + * task becomes ready. Absent for runs that should never wake anyone. + */ + readonly wakeToken?: string; } /** * The durable task run: single writer for one task's lifecycle. * - * Consumes commands over its private hook, applies the pure transition - * function, and appends a full `TaskView` snapshot per accepted command - * to its `eve.task` run stream. Competing completion, cancellation, and - * input transitions serialize here; rejected commands (for example a - * late child result after `cancelled`) change nothing. + * Consumes commands and child wire payloads over its private hook, + * applies the pure transition function, and appends a full `TaskView` + * snapshot per accepted command to its `eve.task` run stream. Competing + * completion, cancellation, and input transitions serialize here; + * rejected commands (for example a late child result after `cancelled`) + * change nothing. + * + * Wake policy: a transition into a ready status — terminal or + * `input_required` — delivers a framework notification to the parent + * session. A parked parent starts a turn; an active turn observes the + * delivery at its next safe boundary. Nothing else wakes the parent. * * The run ends when the task reaches a terminal status. Its snapshot * stream stays readable, so terminal tasks remain peekable; the @@ -30,7 +47,7 @@ export interface TaskRunWorkflowInput { export async function taskRunWorkflow(input: TaskRunWorkflowInput): Promise { "use workflow"; - const commands = createHook({ token: input.commandToken }); + const commands = createHook({ token: input.commandToken }); // The iterator shares the hook's durable cursor; create it before // claiming so conflict replay is consumed by getConflict(), not a // later iterator read. @@ -55,10 +72,16 @@ export async function taskRunWorkflow(input: TaskRunWorkflowInput): Promise[0], + parentSessionId: string, +): ReturnType { + try { + return getSessionTaskIndex(state); + } catch (error) { + logError(log, "failed to read the task index during parent finalization", error, { + parentSessionId, + }); + return []; + } +} diff --git a/packages/eve/src/tasks/json.ts b/packages/eve/src/tasks/json.ts new file mode 100644 index 0000000000..86f769749a --- /dev/null +++ b/packages/eve/src/tasks/json.ts @@ -0,0 +1,43 @@ +import type { JsonObject, JsonValue } from "#shared/json.js"; +import type { TaskView } from "#tasks/types.js"; + +/** + * Projects a task snapshot into the JSON value carried by tool results. + * Field-by-field on purpose: it is the one place that decides what the + * model may see, and it stays a compile error when `TaskView` grows a + * field that needs a disclosure decision. + */ +export function taskViewToJson(view: TaskView): JsonObject { + const metadata: Record = { + kind: view.metadata.kind, + mode: view.metadata.mode, + name: view.metadata.name, + }; + if (view.metadata.childSessionId !== undefined) { + metadata.childSessionId = view.metadata.childSessionId; + } + if (view.metadata.url !== undefined) { + metadata.url = view.metadata.url; + } + + const json: Record = { + metadata, + status: view.status, + taskId: view.taskId, + }; + if (view.statusMessage !== undefined) { + json.statusMessage = view.statusMessage; + } + if (view.lastOutput !== undefined) { + json.lastOutput = { data: view.lastOutput.data, type: view.lastOutput.type }; + } + if (view.inputRequests !== undefined) { + json.inputRequests = [...view.inputRequests]; + } + return json; +} + +/** Projects many snapshots into one `{ tasks }` tool output. */ +export function taskViewsToJson(views: readonly TaskView[]): JsonValue { + return { tasks: views.map((view) => taskViewToJson(view)) }; +} diff --git a/packages/eve/src/tasks/task-id.ts b/packages/eve/src/tasks/task-id.ts index 23f4c58c2b..51c776df10 100644 --- a/packages/eve/src/tasks/task-id.ts +++ b/packages/eve/src/tasks/task-id.ts @@ -1,3 +1,5 @@ +import { createHash } from "node:crypto"; + import { deriveAgentOperationId } from "#harness/handles/operation-id.js"; /** @@ -20,3 +22,23 @@ export function deriveTaskId(input: { }): string { return `task_${deriveAgentOperationId(input).slice(0, 24)}`; } + +/** + * Derives the task run's private command-hook token. + * + * Deterministic on purpose: a durable replay of the dispatch step must + * re-derive the same token so the duplicate task run loses the hook + * claim and exits instead of splitting the lifecycle across two runs. + * Unguessable in practice: the parent session's continuation token is + * itself a private capability, and the derived token never renders to + * the model. + */ +export function deriveTaskCommandToken(input: { + readonly parentContinuationToken: string; + readonly taskId: string; +}): string { + return `task:${input.taskId}:${createHash("sha256") + .update(`${input.taskId}\0${input.parentContinuationToken}`) + .digest("hex") + .slice(0, 32)}`; +} diff --git a/packages/eve/src/tasks/transitions.test.ts b/packages/eve/src/tasks/transitions.test.ts index 0984d2e14c..865c3f2561 100644 --- a/packages/eve/src/tasks/transitions.test.ts +++ b/packages/eve/src/tasks/transitions.test.ts @@ -24,6 +24,7 @@ const ALL_COMMANDS: readonly TaskCommand[] = [ { kind: "cancel" }, { inputRequests: [{ question: "which?" }], kind: "require-input" }, { kind: "resume-working" }, + { childSessionId: "child-session-2", kind: "describe" }, ]; describe("applyTaskTransition", () => { @@ -151,6 +152,25 @@ describe("applyTaskTransition", () => { } }); + it("attaches the child session through describe without changing status", () => { + const described = applyTaskTransition( + createView("working", { + metadata: { kind: "subagent", mode: "local", name: "research" }, + }), + { childSessionId: "child-session-9", kind: "describe" }, + ); + + expect(described.outcome).toBe("accepted"); + expect(described.view.status).toBe("working"); + expect(described.view.metadata.childSessionId).toBe("child-session-9"); + + const again = applyTaskTransition(described.view, { + childSessionId: "child-session-9", + kind: "describe", + }); + expect(again.outcome).toBe("noop"); + }); + it("is deterministic for replayed commands", () => { const view = createView("working"); const command: TaskCommand = { data: { answer: 1 }, kind: "complete" }; diff --git a/packages/eve/src/tasks/transitions.ts b/packages/eve/src/tasks/transitions.ts index 97c5ac0168..7e9a70fdaf 100644 --- a/packages/eve/src/tasks/transitions.ts +++ b/packages/eve/src/tasks/transitions.ts @@ -104,5 +104,18 @@ export function applyTaskTransition(view: TaskView, command: TaskCommand): TaskT }, }; } + case "describe": { + if (view.metadata.childSessionId === command.childSessionId) { + return { outcome: "noop", view }; + } + + return { + outcome: "accepted", + view: { + ...view, + metadata: { ...view.metadata, childSessionId: command.childSessionId }, + }, + }; + } } } diff --git a/packages/eve/src/tasks/types.ts b/packages/eve/src/tasks/types.ts index f8ee790c75..aad27afb37 100644 --- a/packages/eve/src/tasks/types.ts +++ b/packages/eve/src/tasks/types.ts @@ -20,14 +20,22 @@ import type { JsonValue } from "#shared/json.js"; */ export type TaskStatus = "working" | "input_required" | "completed" | "failed" | "cancelled"; -/** Immutable identity of the delegated work behind a task. */ +/** + * Immutable identity of the delegated work behind a task. + * + * `childSessionId` is optional because the task run is created before + * the child acknowledges its session — the durable record must exist + * before the dispatch side effect so a fast child always has a live + * command hook to answer. The `describe` command attaches the id at + * acknowledgement. + */ export interface TaskMetadata { readonly kind: "subagent"; readonly mode: "local" | "remote"; /** Authored subagent name the parent dispatched. */ readonly name: string; /** Child session acknowledged at dispatch. */ - readonly childSessionId: string; + readonly childSessionId?: string; /** Remote children only: the child agent's base URL. */ readonly url?: string; } @@ -73,7 +81,8 @@ export type TaskCommand = | { readonly kind: "fail"; readonly data: JsonValue } | { readonly kind: "cancel" } | { readonly kind: "require-input"; readonly inputRequests: readonly TaskInputRequest[] } - | { readonly kind: "resume-working" }; + | { readonly kind: "resume-working" } + | { readonly kind: "describe"; readonly childSessionId: string }; /** Hook payload envelope commanding a durable task run. */ export interface TaskCommandHookPayload { @@ -81,6 +90,50 @@ export interface TaskCommandHookPayload { readonly command: TaskCommand; } +/** + * Structural shapes of the child wire payloads a task run consumes. + * + * These mirror the existing parent-notification contracts (the local + * `notifyDelegatedParentStep`, the subagent adapter's HITL forwarding, + * and the remote callback route) without importing their zod-backed + * modules: this file is bundled into workflow bodies. The wire itself + * is unchanged — delegated dispatch only points it at the task run's + * hook instead of the parent turn's inbox. + */ +export interface TaskInboundChildResult { + readonly kind: "runtime-action-result"; + readonly results: readonly { + readonly isError?: boolean; + readonly outcome?: { + readonly kind: "parked" | "terminal"; + readonly result: + | { readonly kind: "succeeded"; readonly output: JsonValue } + | { readonly error: JsonValue; readonly kind: "failed" } + | { readonly kind: "cancelled" }; + /** Provider usage this turn added; accounting is deferred to a later stage. */ + readonly usageDelta?: unknown; + }; + readonly output: JsonValue; + }[]; +} + +export interface TaskInboundInputRequest { + readonly kind: "subagent-input-request"; + readonly event: { readonly requests: readonly TaskInputRequest[] }; +} + +export interface TaskInboundAuthorizationEvent { + readonly kind: "subagent-authorization-event"; + readonly event: { readonly type: "authorization.required" | "authorization.completed" }; +} + +/** Everything a task run's command hook may receive. */ +export type TaskRunInboundPayload = + | TaskCommandHookPayload + | TaskInboundChildResult + | TaskInboundInputRequest + | TaskInboundAuthorizationEvent; + /** Namespaced run stream carrying `TaskView` snapshots. */ export const TASK_SNAPSHOT_STREAM_NAMESPACE = "eve.task"; diff --git a/packages/eve/src/tasks/wire.test.ts b/packages/eve/src/tasks/wire.test.ts new file mode 100644 index 0000000000..55f6c8f4a5 --- /dev/null +++ b/packages/eve/src/tasks/wire.test.ts @@ -0,0 +1,105 @@ +import { describe, expect, it } from "vitest"; + +import { translateTaskInboundPayload } from "#tasks/wire.js"; + +const ZERO_USAGE = { cacheReadTokens: 0, cacheWriteTokens: 0, inputTokens: 0, outputTokens: 0 }; + +describe("translateTaskInboundPayload", () => { + it("passes explicit task commands through", () => { + expect( + translateTaskInboundPayload({ command: { kind: "cancel" }, kind: "task-command" }), + ).toEqual({ kind: "cancel" }); + }); + + it("completes on a succeeded child turn outcome, parked or terminal", () => { + for (const kind of ["parked", "terminal"] as const) { + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [ + { + outcome: { + kind, + result: { kind: "succeeded", output: "answer" }, + usageDelta: ZERO_USAGE, + }, + output: "answer", + }, + ], + }), + ).toEqual({ data: "answer", kind: "complete" }); + } + }); + + it("fails on a failed outcome and cancels on a cancelled outcome", () => { + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [ + { + outcome: { + kind: "terminal", + result: { error: { message: "boom" }, kind: "failed" }, + usageDelta: ZERO_USAGE, + }, + output: { message: "boom" }, + }, + ], + }), + ).toEqual({ data: { message: "boom" }, kind: "fail" }); + + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [ + { + outcome: { kind: "terminal", result: { kind: "cancelled" }, usageDelta: ZERO_USAGE }, + output: null, + }, + ], + }), + ).toEqual({ kind: "cancel" }); + }); + + it("falls back to isError when a result carries no outcome", () => { + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [{ isError: true, output: "broken" }], + }), + ).toEqual({ data: "broken", kind: "fail" }); + expect( + translateTaskInboundPayload({ kind: "runtime-action-result", results: [{ output: "ok" }] }), + ).toEqual({ data: "ok", kind: "complete" }); + }); + + it("ignores empty result payloads", () => { + expect( + translateTaskInboundPayload({ kind: "runtime-action-result", results: [] }), + ).toBeUndefined(); + }); + + it("marks the task input_required on a forwarded HITL batch", () => { + expect( + translateTaskInboundPayload({ + event: { requests: [{ prompt: "Which region?" }] }, + kind: "subagent-input-request", + }), + ).toEqual({ inputRequests: [{ prompt: "Which region?" }], kind: "require-input" }); + }); + + it("blocks on authorization.required and resumes on authorization.completed", () => { + expect( + translateTaskInboundPayload({ + event: { type: "authorization.required" }, + kind: "subagent-authorization-event", + }), + ).toEqual({ inputRequests: [{ blockedOn: "authorization" }], kind: "require-input" }); + expect( + translateTaskInboundPayload({ + event: { type: "authorization.completed" }, + kind: "subagent-authorization-event", + }), + ).toEqual({ kind: "resume-working" }); + }); +}); diff --git a/packages/eve/src/tasks/wire.ts b/packages/eve/src/tasks/wire.ts new file mode 100644 index 0000000000..4ab31594b5 --- /dev/null +++ b/packages/eve/src/tasks/wire.ts @@ -0,0 +1,54 @@ +import type { TaskCommand, TaskRunInboundPayload } from "#tasks/types.js"; + +/** + * Translates one inbound hook payload into a lifecycle command. + * + * The child wire is unchanged by `experimental.tasks`; delegated + * dispatch hands children the task run's hook token, so the payloads + * that used to resume the parent turn arrive here instead: + * + * - a settled child turn (local notification or remote callback) + * carries an explicit outcome — its result status decides + * `complete`, `fail`, or `cancel`; + * - a forwarded HITL batch marks the task `input_required` with the + * outstanding requests; + * - `authorization.required` also blocks the task (the child cannot + * proceed without the parent's user), and `authorization.completed` + * returns it to `working`. Authorization payloads never enter the + * snapshot — only the fact that the child is blocked does. + * + * Returns `undefined` for unrecognized payloads, which the run ignores. + */ +export function translateTaskInboundPayload( + payload: TaskRunInboundPayload, +): TaskCommand | undefined { + switch (payload.kind) { + case "task-command": + return payload.command; + case "runtime-action-result": { + const result = payload.results[0]; + if (result === undefined) return undefined; + if (result.outcome !== undefined) { + switch (result.outcome.result.kind) { + case "succeeded": + return { data: result.output, kind: "complete" }; + case "failed": + return { data: result.output, kind: "fail" }; + case "cancelled": + return { kind: "cancel" }; + } + } + return result.isError === true + ? { data: result.output, kind: "fail" } + : { data: result.output, kind: "complete" }; + } + case "subagent-input-request": + return { inputRequests: payload.event.requests, kind: "require-input" }; + case "subagent-authorization-event": + return payload.event.type === "authorization.required" + ? { inputRequests: [{ blockedOn: "authorization" }], kind: "require-input" } + : { kind: "resume-working" }; + default: + return undefined; + } +} From eae4028038f699223bed66f016c89758f2ea2a23 Mon Sep 17 00:00:00 2001 From: Rui Conti Date: Wed, 5 Aug 2026 14:30:40 -0400 Subject: [PATCH 2/2] refactor(eve): split remote subagent startup Signed-off-by: Rui Conti --- .../dispatch-runtime-actions-step.ts | 110 +--------------- .../src/execution/subagent-start-remote.ts | 124 ++++++++++++++++++ 2 files changed, 125 insertions(+), 109 deletions(-) create mode 100644 packages/eve/src/execution/subagent-start-remote.ts diff --git a/packages/eve/src/execution/dispatch-runtime-actions-step.ts b/packages/eve/src/execution/dispatch-runtime-actions-step.ts index bd9e7cd5b3..c75a71f687 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-step.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-step.ts @@ -65,17 +65,13 @@ import { type DurableSessionState, readDurableSession, } from "#execution/durable-session-store.js"; -import { - resolveRemoteAgentForAction, - startRemoteAgentSession, -} from "#execution/remote-agent-dispatch.js"; import { createRecursiveAgentRootOnlyResult, - createRemoteAgentStartFailureResult, createUnavailableDynamicSubagentResult, getSubagentName, } from "#execution/dispatch-action-failures.js"; import { mintStartOperation } from "#execution/dispatch-start-operation.js"; +import { startRemoteSubagent } from "#execution/subagent-start-remote.js"; import { hydrateDurableSession } from "#execution/session.js"; import { buildSubagentRunInput, type SubagentInputSource } from "#execution/subagent-tool.js"; import { createWorkflowRuntime, workflowEntryReference } from "#execution/workflow-runtime.js"; @@ -640,110 +636,6 @@ async function startLocalSubagent(input: { }; } -async function startRemoteSubagent(input: { - readonly action: RuntimeRemoteAgentCallActionRequest; - readonly auth: Parameters[0]["auth"]; - readonly batchEvent: { readonly sequence: number; readonly turnId: string }; - readonly bundle: CompiledBundle; - readonly callbackBaseUrl: string | undefined; - readonly currentSession: RuntimeSession; - readonly dynamicRemoteAgent?: DynamicRemoteAgentConfig; - readonly initiatorAuth: Parameters[0]["initiatorAuth"]; - readonly parentContinuationToken: string | undefined; - readonly persistentSessions: boolean; - readonly session: RuntimeSession; -}): Promise { - const { action } = input; - - // Preflight resolution failures happen before ownership exists, so they - // reject without touching the handle store. - let callbackBaseUrl: string; - let resolvedRemote: ReturnType; - try { - if (input.callbackBaseUrl === undefined) { - throw new Error("Cannot dispatch remote agent without a callback base URL."); - } - callbackBaseUrl = input.callbackBaseUrl; - resolvedRemote = resolveRemoteAgentForAction({ - dynamicRemoteAgent: input.dynamicRemoteAgent, - nodeId: action.nodeId, - remoteAgentName: action.remoteAgentName, - registry: input.bundle.subagentRegistry.subagentsByNodeId, - }); - } catch (error) { - logError(log, "remote agent start failed", error, { - remoteAgentName: action.remoteAgentName, - nodeId: action.nodeId, - callId: action.callId, - }); - return { - kind: "error", - result: createRemoteAgentStartFailureResult({ action, error }), - session: input.currentSession, - }; - } - - const { identity, operation } = mintStartOperation({ - callId: action.callId, - name: action.remoteAgentName, - nodeId: action.nodeId, - parentSessionId: input.session.sessionId, - parentTurnId: input.batchEvent.turnId, - }); - const preparedSession = prepareAgentStart(input.currentSession, { - identity, - operation, - target: { callbackBaseUrl, kind: "agent/remote", url: resolvedRemote.url }, - }); - - try { - const child = await startRemoteAgentSession({ - action, - auth: input.auth, - callbackBaseUrl, - callbackToken: input.parentContinuationToken, - initiatorAuth: input.initiatorAuth, - persistentSessions: input.persistentSessions, - remote: resolvedRemote, - session: input.session, - }); - const address = { - callbackBaseUrl, - kind: "agent/remote", - sessionId: child.sessionId, - url: resolvedRemote.url, - ...(child.continuationToken === undefined - ? {} - : { continuationToken: child.continuationToken }), - } as const; - return { - address, - callId: action.callId, - kind: "called", - name: action.name, - session: confirmAgentStarted(preparedSession, { - address, - operationId: operation.id, - }), - toolName: action.remoteAgentName, - }; - } catch (error) { - logError(log, "remote agent start failed", error, { - remoteAgentName: action.remoteAgentName, - nodeId: action.nodeId, - callId: action.callId, - }); - return { - kind: "error", - result: createRemoteAgentStartFailureResult({ action, error }), - session: rejectAgentEffect(preparedSession, { - disposition: "dead", - operationId: operation.id, - }), - }; - } -} - /** Names one delegated dispatch for its task record, before any child exists. */ function describeDelegatedEntry(entry: Extract): { readonly callId: string; diff --git a/packages/eve/src/execution/subagent-start-remote.ts b/packages/eve/src/execution/subagent-start-remote.ts new file mode 100644 index 0000000000..5669b7a912 --- /dev/null +++ b/packages/eve/src/execution/subagent-start-remote.ts @@ -0,0 +1,124 @@ +import type { DispatchOutcome, RuntimeSession } from "#execution/agent-handle-dispatch.js"; +import { createRemoteAgentStartFailureResult } from "#execution/dispatch-action-failures.js"; +import { mintStartOperation } from "#execution/dispatch-start-operation.js"; +import { + resolveRemoteAgentForAction, + startRemoteAgentSession, +} from "#execution/remote-agent-dispatch.js"; +import { + confirmAgentStarted, + prepareAgentStart, + rejectAgentEffect, +} from "#harness/handles/transitions.js"; +import { createLogger, logError } from "#internal/logging.js"; +import type { RuntimeRemoteAgentCallActionRequest } from "#runtime/actions/types.js"; +import type { CompiledBundle } from "#runtime/sessions/runtime-context-keys.js"; + +const log = createLogger("execution.subagent-start-remote"); + +/** Starts one remote subagent after dispatch planning has selected its target. */ +export async function startRemoteSubagent(input: { + readonly action: RuntimeRemoteAgentCallActionRequest; + readonly auth: Parameters[0]["auth"]; + readonly batchEvent: { readonly sequence: number; readonly turnId: string }; + readonly bundle: CompiledBundle; + readonly callbackBaseUrl: string | undefined; + readonly currentSession: RuntimeSession; + readonly dynamicRemoteAgent?: NonNullable< + Parameters[0]["dynamicRemoteAgent"] + >; + readonly initiatorAuth: Parameters[0]["initiatorAuth"]; + readonly parentContinuationToken: string | undefined; + readonly persistentSessions: boolean; + readonly session: RuntimeSession; +}): Promise { + const { action } = input; + + // Preflight resolution failures happen before ownership exists, so they + // reject without touching the handle store. + let callbackBaseUrl: string; + let resolvedRemote: ReturnType; + try { + if (input.callbackBaseUrl === undefined) { + throw new Error("Cannot dispatch remote agent without a callback base URL."); + } + callbackBaseUrl = input.callbackBaseUrl; + resolvedRemote = resolveRemoteAgentForAction({ + dynamicRemoteAgent: input.dynamicRemoteAgent, + nodeId: action.nodeId, + remoteAgentName: action.remoteAgentName, + registry: input.bundle.subagentRegistry.subagentsByNodeId, + }); + } catch (error) { + logError(log, "remote agent start failed", error, { + remoteAgentName: action.remoteAgentName, + nodeId: action.nodeId, + callId: action.callId, + }); + return { + kind: "error", + result: createRemoteAgentStartFailureResult({ action, error }), + session: input.currentSession, + }; + } + + const { identity, operation } = mintStartOperation({ + callId: action.callId, + name: action.remoteAgentName, + nodeId: action.nodeId, + parentSessionId: input.session.sessionId, + parentTurnId: input.batchEvent.turnId, + }); + const preparedSession = prepareAgentStart(input.currentSession, { + identity, + operation, + target: { callbackBaseUrl, kind: "agent/remote", url: resolvedRemote.url }, + }); + + try { + const child = await startRemoteAgentSession({ + action, + auth: input.auth, + callbackBaseUrl, + callbackToken: input.parentContinuationToken, + initiatorAuth: input.initiatorAuth, + persistentSessions: input.persistentSessions, + remote: resolvedRemote, + session: input.session, + }); + const address = { + callbackBaseUrl, + kind: "agent/remote", + sessionId: child.sessionId, + url: resolvedRemote.url, + ...(child.continuationToken === undefined + ? {} + : { continuationToken: child.continuationToken }), + } as const; + return { + address, + callId: action.callId, + kind: "called", + name: action.name, + session: confirmAgentStarted(preparedSession, { + address, + operationId: operation.id, + }), + toolName: action.remoteAgentName, + }; + } catch (error) { + logError(log, "remote agent start failed", error, { + remoteAgentName: action.remoteAgentName, + nodeId: action.nodeId, + callId: action.callId, + }); + return { + kind: "error", + result: createRemoteAgentStartFailureResult({ action, error }), + session: rejectAgentEffect(preparedSession, { + disposition: "dead", + operationId: operation.id, + }), + }; + } +}