diff --git a/packages/eve/src/execution/dispatch-runtime-actions-step.ts b/packages/eve/src/execution/dispatch-runtime-actions-step.ts index 8964c5362..8c44d25ab 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-step.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-step.ts @@ -31,19 +31,7 @@ import { type RuntimeAgentHandleAction, type RuntimeSession, } from "#execution/agent-handle-dispatch.js"; -import { REMOTE_AGENT_START_FAILED, SUBAGENT_START_FAILED } from "#harness/agent-handle-errors.js"; -import { deriveAgentOperationId } from "#harness/handles/operation-id.js"; -import { - deriveAgentId, - getAgentHandleStore, - type AgentIdentity, - type StartOperation, -} from "#harness/handles/store.js"; -import { - confirmAgentStarted, - prepareAgentStart, - rejectAgentEffect, -} from "#harness/handles/transitions.js"; +import { getAgentHandleStore } from "#harness/handles/store.js"; import { getPendingRuntimeActionBatch } from "#harness/runtime-actions.js"; import { createSubagentCalledEvent, @@ -52,25 +40,28 @@ import { } from "#protocol/message.js"; import type { RuntimeActionRequest, - RuntimeRemoteAgentCallActionRequest, + RuntimeActionResult, RuntimeSubagentCallActionRequest, RuntimeSubagentDispatchFailure, - RuntimeSubagentResult, + RuntimeToolCallActionRequest, } from "#runtime/actions/types.js"; +import { + beginDelegatedTask, + executeTaskControlAction, + failDelegatedDispatch, + isTaskControlAction, + settleDelegatedDispatch, +} from "#execution/tasks/dispatch.js"; import { createDurableSessionState, type DurableSessionState, readDurableSession, } from "#execution/durable-session-store.js"; -import { - resolveRemoteAgentForAction, - startRemoteAgentSession, -} from "#execution/remote-agent-dispatch.js"; import { hydrateDurableSession } from "#execution/session.js"; -import { buildSubagentRunInput, type SubagentInputSource } from "#execution/subagent-tool.js"; -import { createWorkflowRuntime, workflowEntryReference } from "#execution/workflow-runtime.js"; +import type { SubagentInputSource } from "#execution/subagent-tool.js"; +import { startSubagent, type DispatchStartTarget } from "#execution/subagent-start.js"; +import { workflowEntryReference } from "#execution/workflow-runtime.js"; import { createLogger, logError } from "#internal/logging.js"; -import { toErrorMessage } from "#shared/errors.js"; import { readSessionTraceContext } from "#tracing/agent-trace-context-store.js"; import { resolveSubagentDepth } from "#harness/subagent-depth.js"; @@ -83,15 +74,8 @@ type DispatchPlanEntry = readonly agentId: string; } | { readonly kind: "reject"; readonly result: RuntimeSubagentDispatchFailure } - | { readonly kind: "start"; readonly target: DispatchStartTarget }; - -type DispatchStartTarget = - | { - readonly kind: "local"; - readonly action: RuntimeSubagentCallActionRequest; - readonly source: SubagentInputSource; - } - | { readonly kind: "remote"; readonly action: RuntimeRemoteAgentCallActionRequest }; + | { readonly kind: "start"; readonly target: DispatchStartTarget } + | { readonly kind: "task-control"; readonly action: RuntimeToolCallActionRequest }; export async function dispatchRuntimeActionsStep(input: { readonly callbackBaseUrl?: string; @@ -101,7 +85,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"; @@ -132,8 +116,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. @@ -152,7 +140,7 @@ export async function dispatchRuntimeActionsStep(input: { ).length; let nextSession = session; - const results: RuntimeSubagentResult[] = []; + const results: RuntimeActionResult[] = []; try { for (const entry of plan) { @@ -161,6 +149,32 @@ export async function dispatchRuntimeActionsStep(input: { continue; } + if (entry.kind === "task-control") { + const control = await executeTaskControlAction({ + action: entry.action, + bundle, + parentContinuationToken: input.parentContinuationToken, + 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": @@ -169,7 +183,8 @@ export async function dispatchRuntimeActionsStep(input: { agentId: entry.agentId, bundle, currentSession: nextSession, - parentToken: input.parentContinuationToken ?? session.continuationToken, + parentToken: + delegatedParentToken ?? input.parentContinuationToken ?? session.continuationToken, parentTurnId: batch.event.turnId, }); break; @@ -184,7 +199,7 @@ export async function dispatchRuntimeActionsStep(input: { currentSession: nextSession, fanoutSize, initiatorAuth, - parentContinuationToken: input.parentContinuationToken, + parentContinuationToken: delegatedParentToken ?? input.parentContinuationToken, parentTraceContext, persistentSessions, session, @@ -195,10 +210,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.childSessionId, + 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. @@ -264,6 +294,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; @@ -326,309 +360,16 @@ function classifyFreshStart(input: { } } -async function startSubagent(input: { - readonly auth: Parameters[0]["auth"]; - readonly batchEvent: { readonly sequence: number; readonly turnId: string }; - readonly bundle: CompiledBundle; - readonly callbackBaseUrl: string | undefined; - readonly capabilities: Parameters[0]["capabilities"]; - readonly channelMetadata: Parameters[0]["channelMetadata"]; - readonly currentSession: RuntimeSession; - readonly fanoutSize: number; - readonly initiatorAuth: Parameters[0]["initiatorAuth"]; - readonly parentContinuationToken: string | undefined; - readonly parentTraceContext: Parameters[0]["parentTraceContext"]; - readonly persistentSessions: boolean; - readonly session: RuntimeSession; - readonly target: DispatchStartTarget; -}): Promise { - switch (input.target.kind) { - case "local": - return startLocalSubagent({ - action: input.target.action, - auth: input.auth, - batchEvent: input.batchEvent, - bundle: input.bundle, - capabilities: input.capabilities, - channelMetadata: input.channelMetadata, - currentSession: input.currentSession, - fanoutSize: input.fanoutSize, - initiatorAuth: input.initiatorAuth, - parentContinuationToken: input.parentContinuationToken, - parentTraceContext: input.parentTraceContext, - persistentSessions: input.persistentSessions, - session: input.session, - source: input.target.source, - }); - case "remote": - return startRemoteSubagent({ - action: input.target.action, - auth: input.auth, - batchEvent: input.batchEvent, - bundle: input.bundle, - callbackBaseUrl: input.callbackBaseUrl, - currentSession: input.currentSession, - initiatorAuth: input.initiatorAuth, - parentContinuationToken: input.parentContinuationToken, - persistentSessions: input.persistentSessions, - session: input.session, - }); - default: { - const _exhaustive: never = input.target; - return _exhaustive; - } - } -} - -/** - * Mints the deterministic start operation and identity for one dispatch. - * All inputs are parent-controlled, so both exist before the child does - * and a durable replay of the step re-derives the same ownership record. - */ -function mintStartOperation(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; - readonly nodeId: string; - readonly parentSessionId: string; - readonly parentTurnId: string; -}): { readonly identity: AgentIdentity; readonly operation: StartOperation } { - const operationId = deriveAgentOperationId({ - callId: input.callId, - parentSessionId: input.parentSessionId, - parentTurnId: input.parentTurnId, - }); - return { - identity: { - id: deriveAgentId(input.name, operationId), - name: input.name, - nodeId: input.nodeId, - }, - operation: { - callId: input.callId, - id: operationId, - kind: "start", - parentTurnId: input.parentTurnId, - }, - }; -} - -async function startLocalSubagent(input: { - readonly action: RuntimeSubagentCallActionRequest; - readonly auth: Parameters[0]["auth"]; - readonly batchEvent: { readonly sequence: number; readonly turnId: string }; - readonly bundle: CompiledBundle; - readonly capabilities: Parameters[0]["capabilities"]; - readonly channelMetadata: Parameters[0]["channelMetadata"]; - readonly currentSession: RuntimeSession; - readonly fanoutSize: number; - readonly initiatorAuth: Parameters[0]["initiatorAuth"]; - readonly parentContinuationToken: string | undefined; - readonly parentTraceContext: Parameters[0]["parentTraceContext"]; - readonly persistentSessions: boolean; - readonly session: RuntimeSession; - readonly source: SubagentInputSource; -}): Promise { - const { action, source } = input; - const childRuntime = createWorkflowRuntime({ - compiledArtifactsSource: input.bundle.compiledArtifactsSource, - nodeId: action.nodeId, - }); - const { childContinuationToken, runInput } = buildSubagentRunInput({ - action, - auth: input.auth, - batchEvent: input.batchEvent, - capabilities: input.capabilities, - channelMetadata: input.channelMetadata, - fanoutSize: input.fanoutSize, - initiatorAuth: input.initiatorAuth, - parentContinuationToken: input.parentContinuationToken, - parentTraceContext: input.parentTraceContext, - persistentSessions: input.persistentSessions, - session: input.session, - source, - }); - - const targetKind = source.type === "runtime" ? ("agent/self" as const) : ("agent/local" as const); - const { identity, operation } = mintStartOperation({ - callId: action.callId, - name: action.subagentName, - nodeId: action.nodeId, - parentSessionId: input.session.sessionId, - parentTurnId: input.batchEvent.turnId, - }); - // Ownership is recorded before the start side effect, and the prepared - // (or rejected) store rides every outcome into the step result. The - // guarantee is intra-step: a crash between the accepted start and the - // step-result commit still replays the whole dispatch step, so the - // orphan window shrinks to that boundary rather than disappearing. - const preparedSession = prepareAgentStart(input.currentSession, { - identity, - operation, - target: { continuationToken: childContinuationToken, kind: targetKind }, - }); - - let childSessionId: string; - try { - const handle = await childRuntime.run(runInput); - childSessionId = handle.sessionId; - } catch (error) { - logError(log, "local subagent start failed", error, { - callId: action.callId, - nodeId: action.nodeId, - subagentName: action.subagentName, - }); - return { - kind: "error", - result: { - callId: action.callId, - isError: true, - kind: "subagent-result", - output: { - code: SUBAGENT_START_FAILED, - message: toErrorMessage(error), - }, - subagentName: action.subagentName, - }, - session: rejectAgentEffect(preparedSession, { - disposition: "dead", - operationId: operation.id, - }), - }; - } - - return { - callId: action.callId, - childSessionId, - kind: "called", - name: action.name, - session: confirmAgentStarted(preparedSession, { - address: { - continuationToken: childContinuationToken, - kind: targetKind, - sessionId: childSessionId, - }, - operationId: operation.id, - }), - toolName: action.subagentName, - }; -} - -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 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({ - 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, - }); - return { - callId: action.callId, - childSessionId: child.sessionId, - kind: "called", - name: action.name, - remote: { url: resolvedRemote.url }, - session: confirmAgentStarted(preparedSession, { - address: { - callbackBaseUrl, - kind: "agent/remote", - sessionId: child.sessionId, - url: resolvedRemote.url, - ...(child.continuationToken === undefined - ? {} - : { continuationToken: child.continuationToken }), - }, - 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, - }), - }; - } -} - -function createRemoteAgentStartFailureResult(input: { - readonly action: RuntimeRemoteAgentCallActionRequest; - readonly error: unknown; -}): RuntimeSubagentDispatchFailure { - return { - callId: input.action.callId, - isError: true, - kind: "subagent-result", - output: { - code: REMOTE_AGENT_START_FAILED, - message: toErrorMessage(input.error), - }, - subagentName: input.action.remoteAgentName, - }; +} { + 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 createRecursiveAgentRootOnlyResult( 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 a3d4c685b..72c7d0df8 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"; const log = createLogger("execution.dispatch-workflow-runtime-actions"); @@ -32,7 +32,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/subagent-start.ts b/packages/eve/src/execution/subagent-start.ts new file mode 100644 index 000000000..20489dc8a --- /dev/null +++ b/packages/eve/src/execution/subagent-start.ts @@ -0,0 +1,349 @@ +/** + * Fresh subagent starts for the runtime-action dispatch step. + * + * Every start commits an agent handle (`starting`) before its side + * effect and confirms it (`running`) once the child reports + * coordinates, so the returned session owns every child it may have + * created. Split from the dispatch step so plan classification and + * dispatch orchestration stay separate concerns. + */ + +import type { DispatchOutcome, RuntimeSession } from "#execution/agent-handle-dispatch.js"; +import { + resolveRemoteAgentForAction, + startRemoteAgentSession, +} from "#execution/remote-agent-dispatch.js"; +import { buildSubagentRunInput, type SubagentInputSource } from "#execution/subagent-tool.js"; +import { createWorkflowRuntime } from "#execution/workflow-runtime.js"; +import { REMOTE_AGENT_START_FAILED, SUBAGENT_START_FAILED } from "#harness/agent-handle-errors.js"; +import { deriveAgentOperationId } from "#harness/handles/operation-id.js"; +import { deriveAgentId, type AgentIdentity, type StartOperation } from "#harness/handles/store.js"; +import { + confirmAgentStarted, + prepareAgentStart, + rejectAgentEffect, +} from "#harness/handles/transitions.js"; +import { createLogger, logError } from "#internal/logging.js"; +import type { + RuntimeRemoteAgentCallActionRequest, + RuntimeSubagentCallActionRequest, + RuntimeSubagentDispatchFailure, +} from "#runtime/actions/types.js"; +import type { CompiledBundle } from "#runtime/sessions/runtime-context-keys.js"; +import { toErrorMessage } from "#shared/errors.js"; + +const log = createLogger("execution.subagent-start"); + +/** One classified fresh-start target. */ +export type DispatchStartTarget = + | { + readonly kind: "local"; + readonly action: RuntimeSubagentCallActionRequest; + readonly source: SubagentInputSource; + } + | { readonly kind: "remote"; readonly action: RuntimeRemoteAgentCallActionRequest }; + +export async function startSubagent(input: { + readonly auth: Parameters[0]["auth"]; + readonly batchEvent: { readonly sequence: number; readonly turnId: string }; + readonly bundle: CompiledBundle; + readonly callbackBaseUrl: string | undefined; + readonly capabilities: Parameters[0]["capabilities"]; + readonly channelMetadata: Parameters[0]["channelMetadata"]; + readonly currentSession: RuntimeSession; + readonly fanoutSize: number; + readonly initiatorAuth: Parameters[0]["initiatorAuth"]; + readonly parentContinuationToken: string | undefined; + readonly parentTraceContext: Parameters[0]["parentTraceContext"]; + readonly persistentSessions: boolean; + readonly session: RuntimeSession; + readonly target: DispatchStartTarget; +}): Promise { + switch (input.target.kind) { + case "local": + return startLocalSubagent({ + action: input.target.action, + auth: input.auth, + batchEvent: input.batchEvent, + bundle: input.bundle, + capabilities: input.capabilities, + channelMetadata: input.channelMetadata, + currentSession: input.currentSession, + fanoutSize: input.fanoutSize, + initiatorAuth: input.initiatorAuth, + parentContinuationToken: input.parentContinuationToken, + parentTraceContext: input.parentTraceContext, + persistentSessions: input.persistentSessions, + session: input.session, + source: input.target.source, + }); + case "remote": + return startRemoteSubagent({ + action: input.target.action, + auth: input.auth, + batchEvent: input.batchEvent, + bundle: input.bundle, + callbackBaseUrl: input.callbackBaseUrl, + currentSession: input.currentSession, + initiatorAuth: input.initiatorAuth, + parentContinuationToken: input.parentContinuationToken, + persistentSessions: input.persistentSessions, + session: input.session, + }); + default: { + const _exhaustive: never = input.target; + return _exhaustive; + } + } +} + +/** + * Mints the deterministic start operation and identity for one dispatch. + * All inputs are parent-controlled, so both exist before the child does + * and a durable replay of the step re-derives the same ownership record. + */ +function mintStartOperation(input: { + readonly callId: string; + readonly name: string; + readonly nodeId: string; + readonly parentSessionId: string; + readonly parentTurnId: string; +}): { readonly identity: AgentIdentity; readonly operation: StartOperation } { + const operationId = deriveAgentOperationId({ + callId: input.callId, + parentSessionId: input.parentSessionId, + parentTurnId: input.parentTurnId, + }); + return { + identity: { + id: deriveAgentId(input.name, operationId), + name: input.name, + nodeId: input.nodeId, + }, + operation: { + callId: input.callId, + id: operationId, + kind: "start", + parentTurnId: input.parentTurnId, + }, + }; +} + +async function startLocalSubagent(input: { + readonly action: RuntimeSubagentCallActionRequest; + readonly auth: Parameters[0]["auth"]; + readonly batchEvent: { readonly sequence: number; readonly turnId: string }; + readonly bundle: CompiledBundle; + readonly capabilities: Parameters[0]["capabilities"]; + readonly channelMetadata: Parameters[0]["channelMetadata"]; + readonly currentSession: RuntimeSession; + readonly fanoutSize: number; + readonly initiatorAuth: Parameters[0]["initiatorAuth"]; + readonly parentContinuationToken: string | undefined; + readonly parentTraceContext: Parameters[0]["parentTraceContext"]; + readonly persistentSessions: boolean; + readonly session: RuntimeSession; + readonly source: SubagentInputSource; +}): Promise { + const { action, source } = input; + const childRuntime = createWorkflowRuntime({ + compiledArtifactsSource: input.bundle.compiledArtifactsSource, + nodeId: action.nodeId, + }); + const { childContinuationToken, runInput } = buildSubagentRunInput({ + action, + auth: input.auth, + batchEvent: input.batchEvent, + capabilities: input.capabilities, + channelMetadata: input.channelMetadata, + fanoutSize: input.fanoutSize, + initiatorAuth: input.initiatorAuth, + parentContinuationToken: input.parentContinuationToken, + parentTraceContext: input.parentTraceContext, + persistentSessions: input.persistentSessions, + session: input.session, + source, + }); + + const targetKind = source.type === "runtime" ? ("agent/self" as const) : ("agent/local" as const); + const { identity, operation } = mintStartOperation({ + callId: action.callId, + name: action.subagentName, + nodeId: action.nodeId, + parentSessionId: input.session.sessionId, + parentTurnId: input.batchEvent.turnId, + }); + // Ownership is recorded before the start side effect, and the prepared + // (or rejected) store rides every outcome into the step result. The + // guarantee is intra-step: a crash between the accepted start and the + // step-result commit still replays the whole dispatch step, so the + // orphan window shrinks to that boundary rather than disappearing. + const preparedSession = prepareAgentStart(input.currentSession, { + identity, + operation, + target: { continuationToken: childContinuationToken, kind: targetKind }, + }); + + let childSessionId: string; + try { + const handle = await childRuntime.run(runInput); + childSessionId = handle.sessionId; + } catch (error) { + logError(log, "local subagent start failed", error, { + callId: action.callId, + nodeId: action.nodeId, + subagentName: action.subagentName, + }); + return { + kind: "error", + result: { + callId: action.callId, + isError: true, + kind: "subagent-result", + output: { + code: SUBAGENT_START_FAILED, + message: toErrorMessage(error), + }, + subagentName: action.subagentName, + }, + session: rejectAgentEffect(preparedSession, { + disposition: "dead", + operationId: operation.id, + }), + }; + } + + return { + callId: action.callId, + childSessionId, + kind: "called", + name: action.name, + session: confirmAgentStarted(preparedSession, { + address: { + continuationToken: childContinuationToken, + kind: targetKind, + sessionId: childSessionId, + }, + operationId: operation.id, + }), + toolName: action.subagentName, + }; +} + +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 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({ + 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, + }); + return { + callId: action.callId, + childSessionId: child.sessionId, + kind: "called", + name: action.name, + remote: { url: resolvedRemote.url }, + session: confirmAgentStarted(preparedSession, { + address: { + callbackBaseUrl, + kind: "agent/remote", + sessionId: child.sessionId, + url: resolvedRemote.url, + ...(child.continuationToken === undefined + ? {} + : { continuationToken: child.continuationToken }), + }, + 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, + }), + }; + } +} + +function createRemoteAgentStartFailureResult(input: { + readonly action: RuntimeRemoteAgentCallActionRequest; + readonly error: unknown; +}): RuntimeSubagentDispatchFailure { + return { + callId: input.action.callId, + isError: true, + kind: "subagent-result", + output: { + code: REMOTE_AGENT_START_FAILED, + message: toErrorMessage(input.error), + }, + subagentName: input.action.remoteAgentName, + }; +} diff --git a/packages/eve/src/execution/tasks/await-steps.ts b/packages/eve/src/execution/tasks/await-steps.ts new file mode 100644 index 000000000..00be82311 --- /dev/null +++ b/packages/eve/src/execution/tasks/await-steps.ts @@ -0,0 +1,104 @@ +import { + EntityConflictError, + HookNotFoundError, + RunExpiredError, + WorkflowRunNotFoundError, +} from "#compiled/@workflow/errors/index.js"; + +import type { AwaitedTaskRef } from "#execution/tasks/await-workflow.js"; +import { readLatestTaskSnapshot } from "#execution/tasks/run-control.js"; +import { getHookByToken } from "#internal/workflow/runtime.js"; +import { createLogger } from "#internal/logging.js"; +import type { RuntimeActionResult } from "#runtime/actions/types.js"; +import { walkCauseChain } from "#shared/errors.js"; +import { taskViewsToJson } from "#tasks/json.js"; +import type { TaskView } from "#tasks/types.js"; +import { resumeHook } from "#internal/workflow/runtime.js"; + +const log = createLogger("execution.tasks.await"); + +/** + * Reads the latest snapshot of every awaited task, or reports that the + * waiting turn's inbox is gone so the aggregation run can stop polling. + * + * A run that has not published its first snapshot yet reads as + * `working` — the caller holds the creation receipt, which says the + * same thing. + */ +export async function readAwaitedTaskViewsStep(input: { + readonly replyToken: string; + readonly tasks: readonly AwaitedTaskRef[]; +}): Promise< + { readonly kind: "listener-gone" } | { readonly kind: "views"; readonly views: TaskView[] } +> { + "use step"; + + try { + await getHookByToken(input.replyToken); + } catch (error) { + if (isGoneListener(error)) { + return { kind: "listener-gone" }; + } + throw error; + } + + const views = await Promise.all( + input.tasks.map( + async (task) => + (await readLatestTaskSnapshot({ taskRunId: task.taskRunId })) ?? + createPendingView(task.taskId), + ), + ); + return { kind: "views", views }; +} + +/** Resolves the pending `task_await` key with its aggregated views. */ +export async function postTaskAwaitResultStep(input: { + readonly callId: string; + readonly replyToken: string; + readonly toolName: string; + readonly views: readonly TaskView[]; +}): Promise { + "use step"; + + const result: RuntimeActionResult = { + callId: input.callId, + kind: "tool-result", + output: taskViewsToJson(input.views), + toolName: input.toolName, + }; + try { + await resumeHook(input.replyToken, { kind: "runtime-action-result", results: [result] }); + } catch (error) { + if (isGoneListener(error)) { + log.warn("task_await listener disappeared before its result posted", { + callId: input.callId, + toolName: input.toolName, + }); + return; + } + throw error; + } +} + +function createPendingView(taskId: string): TaskView { + return { + metadata: { kind: "subagent", mode: "local", name: "unknown" }, + status: "working", + taskId, + }; +} + +function isGoneListener(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/await-workflow.test.ts b/packages/eve/src/execution/tasks/await-workflow.test.ts new file mode 100644 index 000000000..0b6a88344 --- /dev/null +++ b/packages/eve/src/execution/tasks/await-workflow.test.ts @@ -0,0 +1,71 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { sleep } from "#compiled/@workflow/core/index.js"; + +import { postTaskAwaitResultStep, readAwaitedTaskViewsStep } from "#execution/tasks/await-steps.js"; +import { taskAwaitWorkflow } from "#execution/tasks/await-workflow.js"; +import type { TaskView } from "#tasks/types.js"; + +vi.mock("#compiled/@workflow/core/index.js", () => ({ + sleep: vi.fn(), +})); + +vi.mock("./await-steps.js", () => ({ + postTaskAwaitResultStep: vi.fn(), + readAwaitedTaskViewsStep: vi.fn(), +})); + +afterEach(() => { + vi.resetAllMocks(); +}); + +function createView(taskId: string, status: TaskView["status"]): TaskView { + return { + metadata: { kind: "subagent", mode: "local", name: "research" }, + status, + taskId, + }; +} + +const INPUT = { + callId: "call-await-1", + replyToken: "turn-inbox-token", + tasks: [ + { taskId: "task_a", taskRunId: "run-a" }, + { taskId: "task_b", taskRunId: "run-b" }, + ], + toolName: "task_await", +}; + +describe("taskAwaitWorkflow", () => { + it("polls until every task is ready, then posts one aggregated result", async () => { + vi.mocked(readAwaitedTaskViewsStep) + .mockResolvedValueOnce({ + kind: "views", + views: [createView("task_a", "completed"), createView("task_b", "working")], + }) + .mockResolvedValueOnce({ + kind: "views", + views: [createView("task_a", "completed"), createView("task_b", "input_required")], + }); + vi.mocked(sleep).mockResolvedValue(undefined); + + await taskAwaitWorkflow(INPUT); + + expect(sleep).toHaveBeenCalledTimes(1); + expect(postTaskAwaitResultStep).toHaveBeenCalledTimes(1); + expect(vi.mocked(postTaskAwaitResultStep).mock.calls[0]?.[0]).toMatchObject({ + callId: "call-await-1", + replyToken: "turn-inbox-token", + toolName: "task_await", + }); + }); + + it("stops polling without posting when the waiting turn is gone", async () => { + vi.mocked(readAwaitedTaskViewsStep).mockResolvedValue({ kind: "listener-gone" }); + + await taskAwaitWorkflow(INPUT); + + expect(postTaskAwaitResultStep).not.toHaveBeenCalled(); + expect(sleep).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/eve/src/execution/tasks/await-workflow.ts b/packages/eve/src/execution/tasks/await-workflow.ts new file mode 100644 index 000000000..a8376ff93 --- /dev/null +++ b/packages/eve/src/execution/tasks/await-workflow.ts @@ -0,0 +1,60 @@ +import { sleep } from "#compiled/@workflow/core/index.js"; + +import { postTaskAwaitResultStep, readAwaitedTaskViewsStep } from "#execution/tasks/await-steps.js"; +import { isReadyTaskStatus } from "#tasks/types.js"; + +const DEFAULT_POLL_INTERVAL_MS = 10_000; + +/** One awaited task: the model-visible id plus its run's read coordinates. */ +export interface AwaitedTaskRef { + readonly taskId: string; + readonly taskRunId: string; +} + +/** Input for one `task_await` aggregation run. */ +export interface TaskAwaitWorkflowInput { + readonly callId: string; + readonly pollIntervalMs?: number; + /** The waiting turn's inbox token; the result resumes the existing wait. */ + readonly replyToken: string; + readonly tasks: readonly AwaitedTaskRef[]; + readonly toolName: string; +} + +/** + * Aggregates one `task_await` call across its selected task runs. + * + * `task_await` returns when *every* selected task is terminal or + * `input_required`, so someone must observe all of them; the task runs + * are single-task writers and the parent turn can only wait on its + * inbox. This small durable run polls the snapshot streams and, once + * every task is ready, posts the one `tool-result` the pending + * `task_await` key is waiting for. + * + * The run exits without posting when the waiting turn is gone (its + * inbox was disposed by completion or cancellation) — the model asked, + * then stopped listening. + */ +export async function taskAwaitWorkflow(input: TaskAwaitWorkflowInput): Promise { + "use workflow"; + + while (true) { + const observation = await readAwaitedTaskViewsStep({ + replyToken: input.replyToken, + tasks: input.tasks, + }); + if (observation.kind === "listener-gone") return; + + if (observation.views.every((view) => isReadyTaskStatus(view.status))) { + await postTaskAwaitResultStep({ + callId: input.callId, + replyToken: input.replyToken, + toolName: input.toolName, + views: observation.views, + }); + return; + } + + await sleep(input.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS); + } +} diff --git a/packages/eve/src/execution/tasks/dispatch.ts b/packages/eve/src/execution/tasks/dispatch.ts new file mode 100644 index 000000000..82b29a241 --- /dev/null +++ b/packages/eve/src/execution/tasks/dispatch.ts @@ -0,0 +1,365 @@ +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 type { AwaitedTaskRef } from "#execution/tasks/await-workflow.js"; +import { + requestWorkflowTurnCancellation, + startWorkflowPreferLatest, + taskAwaitWorkflowReference, +} 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_AWAIT_TOOL_NAME, + 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_await` / `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", + 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, + sessionId: input.childSessionId, + 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. `task_await` is the exception: when any selected task is + * still working it starts the aggregation run and returns no result, + * leaving the pending key to the turn's existing inbox wait. + */ +export async function executeTaskControlAction(input: { + readonly action: RuntimeToolCallActionRequest; + readonly bundle: CompiledBundle; + readonly parentContinuationToken: string | undefined; + 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) }; + } + case TASK_AWAIT_TOOL_NAME: { + const views = await readTaskViews(entries); + if (views.every((view) => isReadyTaskStatus(view.status))) { + return { result: createTaskViewsResult(action, views) }; + } + if (input.parentContinuationToken === undefined) { + return { + result: createTaskControlError( + action, + "task_await is unavailable on this session driver.", + ), + }; + } + const tasks: AwaitedTaskRef[] = entries.map((entry) => ({ + taskId: entry.taskId, + taskRunId: entry.taskRunId, + })); + await startWorkflowPreferLatest(taskAwaitWorkflowReference, [ + { + callId: action.callId, + replyToken: input.parentContinuationToken, + tasks, + toolName: action.toolName, + }, + ]); + return { result: undefined }; + } + 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 d3e00912c..5e14ac126 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 faa2ebbb6..0c8c50af1 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 f5725eef4..f540f071e 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 b8df99788..4d9eaf6a4 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/execution/workflow-runtime.ts b/packages/eve/src/execution/workflow-runtime.ts index 2dae778b4..081dbf541 100644 --- a/packages/eve/src/execution/workflow-runtime.ts +++ b/packages/eve/src/execution/workflow-runtime.ts @@ -57,6 +57,7 @@ const WORKFLOW_ENTRY_NAME = "workflowEntry"; const TURN_WORKFLOW_NAME = "turnWorkflow"; const SESSION_TIMEOUT_WORKFLOW_NAME = "sessionTimeoutWorkflow"; const TASK_RUN_WORKFLOW_NAME = "taskRunWorkflow"; +const TASK_AWAIT_WORKFLOW_NAME = "taskAwaitWorkflow"; const EVE_PACKAGE_INFO = resolveInstalledPackageInfo(); export const LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE = @@ -77,6 +78,7 @@ export const STABLE_WORKFLOW_NAMES: ReadonlySet = new Set([ TURN_WORKFLOW_NAME, SESSION_TIMEOUT_WORKFLOW_NAME, TASK_RUN_WORKFLOW_NAME, + TASK_AWAIT_WORKFLOW_NAME, ]); const STABLE_ID_BASE = EVE_PACKAGE_INFO.name; @@ -118,6 +120,11 @@ export const taskRunWorkflowReference = { workflowId: `workflow//${STABLE_ID_BASE}//${TASK_RUN_WORKFLOW_NAME}`, }; +/** Stable workflow reference for `task_await` aggregation runs. */ +export const taskAwaitWorkflowReference = { + workflowId: `workflow//${STABLE_ID_BASE}//${TASK_AWAIT_WORKFLOW_NAME}`, +}; + /** * Creates a workflow-backed runtime whose long-lived driver owns the * event stream and dispatches each turn as a child workflow run. diff --git a/packages/eve/src/tasks/json.ts b/packages/eve/src/tasks/json.ts new file mode 100644 index 000000000..86f769749 --- /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 23f4c58c2..51c776df1 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 0984d2e14..865c3f256 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 97c5ac016..7e9a70fda 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 b339accdc..c2887ff80 100644 --- a/packages/eve/src/tasks/types.ts +++ b/packages/eve/src/tasks/types.ts @@ -21,14 +21,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; } @@ -74,7 +82,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 { @@ -82,6 +91,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 000000000..55f6c8f4a --- /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 000000000..4ab31594b --- /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; + } +}