diff --git a/.changeset/tasks-experimental-subagents.md b/.changeset/tasks-experimental-subagents.md new file mode 100644 index 000000000..066b48883 --- /dev/null +++ b/.changeset/tasks-experimental-subagents.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Add experimental background tasks for subagents. With `experimental.tasks` on the root agent, subagent calls return a task receipt immediately instead of blocking the turn, and the model manages the delegated work with the new `task_peek`, `task_cancel`, `task_send`, and `task_sleep` tools. Terminal results and input requests wake the parent through the normal session delivery path. Without the flag, nothing changes. diff --git a/packages/eve/src/cli/dev/tui/tool-presentation.test.ts b/packages/eve/src/cli/dev/tui/tool-presentation.test.ts index 246275e54..1de393e84 100644 --- a/packages/eve/src/cli/dev/tui/tool-presentation.test.ts +++ b/packages/eve/src/cli/dev/tui/tool-presentation.test.ts @@ -180,6 +180,7 @@ describe("presentTool", () => { read_file: { filePath: "/workspace/a.ts" }, task_cancel: { taskIds: ["task_abc"] }, task_peek: { taskIds: ["task_abc"] }, + task_send: { message: "Continue with the next region.", taskId: "task_abc" }, task_sleep: { seconds: 30 }, todo: { todos: [] }, web_fetch: { url: "https://example.com" }, diff --git a/packages/eve/src/cli/dev/tui/tool-presentation.ts b/packages/eve/src/cli/dev/tui/tool-presentation.ts index b2a94f836..a27542e24 100644 --- a/packages/eve/src/cli/dev/tui/tool-presentation.ts +++ b/packages/eve/src/cli/dev/tui/tool-presentation.ts @@ -139,6 +139,13 @@ const BUILTIN_TOOL_COPY: Readonly> = { singularNoun: "task", pluralNoun: "tasks", }, + task_send: { + verb: "Send", + pastVerb: "Sent", + argKey: "taskId", + singularNoun: "task", + pluralNoun: "tasks", + }, task_sleep: { verb: "Pause", pastVerb: "Paused", diff --git a/packages/eve/src/execution/dispatch-runtime-actions-step.ts b/packages/eve/src/execution/dispatch-runtime-actions-step.ts index c75a71f68..8015d541e 100644 --- a/packages/eve/src/execution/dispatch-runtime-actions-step.ts +++ b/packages/eve/src/execution/dispatch-runtime-actions-step.ts @@ -199,8 +199,10 @@ export async function dispatchRuntimeActionsStep(input: { const control = await executeTaskControlAction({ action: entry.action, bundle, + parentTurnId: batch.event.turnId, session: nextSession, }); + nextSession = control.session; if (control.result !== undefined) { results.push(control.result); } diff --git a/packages/eve/src/execution/node-step.test.ts b/packages/eve/src/execution/node-step.test.ts index 52bc39e7d..77dc06da4 100644 --- a/packages/eve/src/execution/node-step.test.ts +++ b/packages/eve/src/execution/node-step.test.ts @@ -276,7 +276,7 @@ describe("createNodeHarnessTools", () => { it("does not inject task tools without experimental.tasks", () => { const tools = createNodeHarnessTools({ node: createTestNode() }); - for (const name of ["task_peek", "task_cancel", "task_sleep"]) { + for (const name of ["task_peek", "task_cancel", "task_send", "task_sleep"]) { expect(tools.has(name)).toBe(false); } }); @@ -293,7 +293,7 @@ describe("createNodeHarnessTools", () => { }, }); - for (const name of ["task_peek", "task_cancel"]) { + for (const name of ["task_peek", "task_cancel", "task_send"]) { expect(tools.get(name)?.runtimeAction).toEqual({ kind: "task-control" }); expect(tools.get(name)?.execute).toBeUndefined(); } diff --git a/packages/eve/src/execution/tasks/control-shared.ts b/packages/eve/src/execution/tasks/control-shared.ts new file mode 100644 index 000000000..6602c5582 --- /dev/null +++ b/packages/eve/src/execution/tasks/control-shared.ts @@ -0,0 +1,108 @@ +import type { RuntimeSession } from "#execution/agent-handle-dispatch.js"; +import { readLatestTaskSnapshot } from "#execution/tasks/run-control.js"; +import { getAgentHandleStore, type AgentHandle } from "#harness/handles/store.js"; +import type { RuntimeActionResult, RuntimeToolCallActionRequest } from "#runtime/actions/types.js"; +import { taskViewsToJson } from "#tasks/json.js"; +import { findSessionTaskEntry, type SessionTaskIndexEntry } from "#tasks/session-index.js"; +import type { TaskView } from "#tasks/types.js"; + +/** + * Result and lookup helpers shared by the task-control executors + * (`task_peek`/`task_cancel` in the dispatch module, + * `task_send` in its own). + */ + +/** Resolves owned index entries, or the ids this session does not own. */ +export function lookupTaskEntries( + session: RuntimeSession, + taskIds: readonly string[], +): + | { readonly entries: SessionTaskIndexEntry[]; readonly kind: "found" } + | { readonly kind: "unknown"; readonly unknown: string[] } { + const entries: SessionTaskIndexEntry[] = []; + const unknown: string[] = []; + for (const taskId of taskIds) { + const entry = findSessionTaskEntry(session.state, taskId); + if (entry === undefined) { + unknown.push(taskId); + } else { + entries.push(entry); + } + } + return unknown.length > 0 ? { kind: "unknown", unknown } : { entries, kind: "found" }; +} + +/** Reads the latest snapshot of every entry, defaulting to `working`. */ +export async function readTaskViews( + entries: readonly SessionTaskIndexEntry[], +): Promise { + return Promise.all( + entries.map( + async (entry) => + (await readLatestTaskSnapshot({ taskRunId: entry.taskRunId })) ?? + createPendingTaskView(entry.taskId), + ), + ); +} + +/** The view of a run that has not published its first snapshot yet. */ +export function createPendingTaskView(taskId: string): TaskView { + return { + metadata: { kind: "subagent", mode: "local", name: "unknown" }, + status: "working", + taskId, + }; +} + +/** Finds the handle owning one child session's address, any live phase. */ +export function findAddressableHandle( + session: RuntimeSession, + childSessionId: string | undefined, +): Extract | undefined { + if (childSessionId === undefined) return undefined; + const handles = getAgentHandleStore(session.state)?.handles ?? []; + return handles + .filter( + (candidate): candidate is Extract => + candidate.phase === "running" || candidate.phase === "parked", + ) + .find((candidate) => candidate.address.sessionId === childSessionId); +} + +/** One successful task-control result carrying full task views. */ +export function createTaskViewsResult( + action: RuntimeToolCallActionRequest, + views: readonly TaskView[], +): RuntimeActionResult { + return { + callId: action.callId, + kind: "tool-result", + output: taskViewsToJson(views), + toolName: action.toolName, + }; +} + +/** One task-control error the model can act on. */ +export function createTaskControlError( + action: RuntimeToolCallActionRequest, + message: string, +): RuntimeActionResult { + return { + callId: action.callId, + isError: true, + kind: "tool-result", + output: { message }, + toolName: action.toolName, + }; +} + +/** The ownership error for ids outside this session's task index. */ +export function createUnknownTasksError( + action: RuntimeToolCallActionRequest, + unknown: readonly string[], +): RuntimeActionResult { + return createTaskControlError( + action, + `Unknown task ids: ${unknown.join(", ")}. Tasks belong to the session that created them.`, + ); +} diff --git a/packages/eve/src/execution/tasks/delegate.ts b/packages/eve/src/execution/tasks/delegate.ts new file mode 100644 index 000000000..cfb7b4c31 --- /dev/null +++ b/packages/eve/src/execution/tasks/delegate.ts @@ -0,0 +1,113 @@ +import type { RuntimeSession } from "#execution/agent-handle-dispatch.js"; +import { sendTaskCommand, startTaskRun } from "#execution/tasks/run-control.js"; +import type { RuntimeSubagentChildResult } from "#runtime/actions/types.js"; +import type { JsonValue } from "#shared/json.js"; +import { recordSessionTask } from "#tasks/session-index.js"; +import { deriveTaskCommandToken, deriveTaskId } from "#tasks/task-id.js"; + +/** A prepared delegated task: identity plus its started durable run. */ +export interface DelegatedTask { + readonly commandToken: string; + readonly taskId: string; + readonly taskRunId: string; +} + +/** + * 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 }, + }); +} diff --git a/packages/eve/src/execution/tasks/dispatch.ts b/packages/eve/src/execution/tasks/dispatch.ts index f3a1858ea..f7d445511 100644 --- a/packages/eve/src/execution/tasks/dispatch.ts +++ b/packages/eve/src/execution/tasks/dispatch.ts @@ -4,17 +4,21 @@ import { resolveRemoteAgentForAction, } from "#execution/remote-agent-dispatch.js"; import { - readLatestTaskSnapshot, - sendTaskCommand, - startTaskRun, -} from "#execution/tasks/run-control.js"; + createPendingTaskView, + createTaskControlError, + createTaskViewsResult, + createUnknownTasksError, + findAddressableHandle, + lookupTaskEntries, + readTaskViews, +} from "#execution/tasks/control-shared.js"; +import { executeTaskSend } from "#execution/tasks/send.js"; +import { readLatestTaskSnapshot, sendTaskCommand } 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"; @@ -22,189 +26,82 @@ import { TASK_CANCEL_TOOL_NAME, TASK_CONTROL_TOOL_NAMES, TASK_PEEK_TOOL_NAME, + TASK_SEND_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 type { SessionTaskIndexEntry } from "#tasks/session-index.js"; import { isReadyTaskStatus, type TaskView } from "#tasks/types.js"; +export { + beginDelegatedTask, + failDelegatedDispatch, + settleDelegatedDispatch, + type DelegatedTask, +} from "#execution/tasks/delegate.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. */ +/** True for `task_peek` / `task_cancel` / `task_send` 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. + * + * Returns the (possibly updated) session: `task_send` follow-ups record + * new tasks and settle the continued agent handle. */ export async function executeTaskControlAction(input: { readonly action: RuntimeToolCallActionRequest; readonly bundle: CompiledBundle; + readonly parentTurnId: string; readonly session: RuntimeSession; -}): Promise<{ readonly result: RuntimeActionResult | undefined }> { - const { action } = input; +}): Promise<{ + readonly result: RuntimeActionResult | undefined; + readonly session: RuntimeSession; +}> { + const { action, session } = input; + + if (action.toolName === TASK_SEND_TOOL_NAME) { + return executeTaskSend(input); + } + const taskIds = readTaskIds(action.input); if (taskIds === undefined || taskIds.length === 0) { return { result: createTaskControlError(action, "Provide a non-empty `taskIds` array."), + session, }; } - 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.`, - ), - }; + const lookup = lookupTaskEntries(session, taskIds); + if (lookup.kind === "unknown") { + return { result: createUnknownTasksError(action, lookup.unknown), session }; } + const entries = lookup.entries; switch (action.toolName) { case TASK_PEEK_TOOL_NAME: { const views = await readTaskViews(entries); - return { result: createTaskViewsResult(action, views) }; + return { result: createTaskViewsResult(action, views), session }; } case TASK_CANCEL_TOOL_NAME: { const views = await Promise.all( - entries.map((entry) => - cancelOneTask({ bundle: input.bundle, entry, session: input.session }), - ), + entries.map((entry) => cancelOneTask({ bundle: input.bundle, entry, session })), ); - return { result: createTaskViewsResult(action, views) }; + return { result: createTaskViewsResult(action, views), session }; } default: return { result: createTaskControlError(action, `Unsupported task control "${action.toolName}".`), + session, }; } } @@ -249,13 +146,7 @@ async function propagateTaskCancel(input: { }): 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); + const handle = findAddressableHandle(input.session, childSessionId); try { if (handle !== undefined && handle.address.kind === "agent/remote") { @@ -279,49 +170,6 @@ async function propagateTaskCancel(input: { } } -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; diff --git a/packages/eve/src/execution/tasks/send.ts b/packages/eve/src/execution/tasks/send.ts new file mode 100644 index 000000000..e541bc7bc --- /dev/null +++ b/packages/eve/src/execution/tasks/send.ts @@ -0,0 +1,326 @@ +import { + dispatchToAgentHandle, + type RuntimeAgentHandleAction, + type RuntimeSession, +} from "#execution/agent-handle-dispatch.js"; +import { + createPendingTaskView, + createTaskControlError, + createTaskViewsResult, + createUnknownTasksError, + findAddressableHandle, +} from "#execution/tasks/control-shared.js"; +import { + beginDelegatedTask, + failDelegatedDispatch, + settleDelegatedDispatch, +} from "#execution/tasks/delegate.js"; +import { readLatestTaskSnapshot, sendTaskCommand } from "#execution/tasks/run-control.js"; +import { createWorkflowRuntime } from "#execution/workflow-runtime.js"; +import { AGENT_BUSY, AGENT_UNREACHABLE } from "#harness/agent-handle-errors.js"; +import { deriveAgentOperationId } from "#harness/handles/operation-id.js"; +import { settleAgentTurn } from "#harness/handles/transitions.js"; +import { createLogger, logError } from "#internal/logging.js"; +import type { RuntimeActionResult, RuntimeToolCallActionRequest } from "#runtime/actions/types.js"; +import type { CompiledBundle } from "#runtime/sessions/runtime-context-keys.js"; +import { findSessionTaskEntry, type SessionTaskIndexEntry } from "#tasks/session-index.js"; +import { applyTaskTransition } from "#tasks/transitions.js"; +import type { TaskView } from "#tasks/types.js"; + +const log = createLogger("execution.tasks.send"); + +/** + * Routes one `task_send`: + * + * - `working` tasks are busy — the send surfaces `AGENT_BUSY` instead + * of queueing (settled decision; queueing is the reversible follow-up); + * - `input_required` tasks accept an `inputResponses` batch, delivered + * to the parked child session, and return to `working`; + * - terminal tasks accept a `message` follow-up, which starts a new + * task bound to the same child session and returns its receipt. + */ +export async function executeTaskSend(input: { + readonly action: RuntimeToolCallActionRequest; + readonly bundle: CompiledBundle; + readonly parentTurnId: string; + readonly session: RuntimeSession; +}): Promise<{ + readonly result: RuntimeActionResult | undefined; + readonly session: RuntimeSession; +}> { + const { action, session } = input; + const send = readTaskSendInput(action.input); + if (send.kind === "invalid") { + return { result: createTaskControlError(action, send.message), session }; + } + + const entry = findSessionTaskEntry(session.state, send.taskId); + if (entry === undefined) { + return { result: createUnknownTasksError(action, [send.taskId]), session }; + } + + const view = + (await readLatestTaskSnapshot({ taskRunId: entry.taskRunId })) ?? + createPendingTaskView(entry.taskId); + + if (view.status === "working") { + return { + result: createTaskControlError( + action, + `${AGENT_BUSY}: task "${view.taskId}" is still working. Let it finish, or cancel it first.`, + ), + session, + }; + } + + if (view.status === "input_required") { + if (send.body.kind !== "input-responses") { + return { + result: createTaskControlError( + action, + `Task "${view.taskId}" is waiting on input; answer it with inputResponses.`, + ), + session, + }; + } + return answerBlockedTask({ + action, + bundle: input.bundle, + entry, + responses: send.body.inputResponses, + session, + view, + }); + } + + if (send.body.kind !== "message") { + return { + result: createTaskControlError( + action, + `Task "${view.taskId}" is ${view.status}; send a follow-up message to continue its agent.`, + ), + session, + }; + } + return followUpTerminalTask({ + action, + bundle: input.bundle, + message: send.body.message, + parentTurnId: input.parentTurnId, + session, + view, + }); +} + +async function answerBlockedTask(input: { + readonly action: RuntimeToolCallActionRequest; + readonly bundle: CompiledBundle; + readonly entry: SessionTaskIndexEntry; + readonly responses: readonly { readonly requestId: string }[]; + readonly session: RuntimeSession; + readonly view: TaskView; +}): Promise<{ readonly result: RuntimeActionResult; readonly session: RuntimeSession }> { + const { action, session, view } = input; + const handle = findAddressableHandle(session, view.metadata.childSessionId); + if (handle === undefined || handle.address.kind === "agent/remote") { + return { + result: createTaskControlError( + action, + `${AGENT_UNREACHABLE}: task "${view.taskId}" has no reachable child session for input responses.`, + ), + session, + }; + } + + const childRuntime = createWorkflowRuntime({ + compiledArtifactsSource: input.bundle.compiledArtifactsSource, + nodeId: handle.identity.nodeId, + }); + try { + // The child parked waiting on this batch; its next settled turn + // reports to the same task run through the caller reply token. + const result = await childRuntime.dispatchSession({ + command: { + caller: { + callId: action.callId, + replyTo: { kind: "hook", token: input.entry.commandToken }, + subagentName: handle.identity.name, + }, + kind: "send", + payload: { inputResponses: [...input.responses] }, + }, + sessionId: handle.address.sessionId, + }); + if (result.status === "session_not_active") { + throw new Error(`Agent session "${handle.address.sessionId}" is no longer active.`); + } + } catch (error) { + logError(log, "task_send input-response delivery failed", error, { + childSessionId: handle.address.sessionId, + taskId: view.taskId, + }); + return { + result: createTaskControlError( + action, + `${AGENT_UNREACHABLE}: task "${view.taskId}"'s child session did not accept the responses.`, + ), + session, + }; + } + + await sendTaskCommand({ + command: { kind: "resume-working" }, + commandToken: input.entry.commandToken, + }); + const resumed = applyTaskTransition(view, { kind: "resume-working" }); + return { result: createTaskViewsResult(action, [resumed.view]), session }; +} + +async function followUpTerminalTask(input: { + readonly action: RuntimeToolCallActionRequest; + readonly bundle: CompiledBundle; + readonly message: string; + readonly parentTurnId: string; + readonly session: RuntimeSession; + readonly view: TaskView; +}): Promise<{ readonly result: RuntimeActionResult; readonly session: RuntimeSession }> { + const { action, view } = input; + const handle = findAddressableHandle(input.session, view.metadata.childSessionId); + if (handle === undefined) { + return { + result: createTaskControlError( + action, + `${AGENT_UNREACHABLE}: task "${view.taskId}"'s agent is no longer addressable.`, + ), + session: input.session, + }; + } + + const continuation: RuntimeAgentHandleAction = + handle.address.kind === "agent/remote" + ? { + callId: action.callId, + description: "", + input: { message: input.message }, + kind: "remote-agent-call", + name: handle.identity.name, + nodeId: handle.identity.nodeId, + remoteAgentName: handle.identity.name, + } + : { + callId: action.callId, + description: "", + input: { message: input.message }, + kind: "subagent-call", + name: handle.identity.name, + nodeId: handle.identity.nodeId, + subagentName: handle.identity.name, + }; + + const task = await beginDelegatedTask({ + callId: action.callId, + mode: handle.address.kind === "agent/remote" ? "remote" : "local", + name: handle.identity.name, + parentSessionId: input.session.sessionId, + parentTurnId: input.parentTurnId, + session: input.session, + }); + const outcome = await dispatchToAgentHandle({ + action: continuation, + agentId: handle.identity.id, + bundle: input.bundle, + currentSession: input.session, + parentToken: task.commandToken, + parentTurnId: input.parentTurnId, + }); + if (outcome.kind === "error") { + await failDelegatedDispatch({ error: outcome.result.output, task }); + return { + result: { + callId: action.callId, + isError: true, + kind: "tool-result", + output: outcome.result.output, + toolName: action.toolName, + }, + session: outcome.session, + }; + } + + const settled = await settleDelegatedDispatch({ + callId: outcome.callId, + childSessionId: outcome.address.sessionId, + session: outcome.session, + subagentName: outcome.toolName, + task, + }); + // task_send's own result is a tool-result, so the receipt never flows + // through the handle-settling resolve path; park the continued handle + // here to keep it addressable for later sends. + const operationId = deriveAgentOperationId({ + callId: action.callId, + parentSessionId: input.session.sessionId, + parentTurnId: input.parentTurnId, + }); + const settledHandle = settleAgentTurn(settled.session, { + operationId, + outcome: { + kind: "parked", + result: { + kind: "succeeded", + output: `Delegated as background task ${task.taskId} (working).`, + }, + usageDelta: { cacheReadTokens: 0, cacheWriteTokens: 0, inputTokens: 0, outputTokens: 0 }, + }, + }); + return { + result: { + callId: action.callId, + kind: "tool-result", + output: { status: "working", taskId: task.taskId }, + toolName: action.toolName, + }, + session: settledHandle.kind === "settled" ? settledHandle.session : settled.session, + }; +} + +type TaskSendInput = + | { readonly kind: "invalid"; readonly message: string } + | { + readonly body: + | { readonly kind: "message"; readonly message: string } + | { + readonly inputResponses: readonly { readonly requestId: string }[]; + readonly kind: "input-responses"; + }; + readonly kind: "send"; + readonly taskId: string; + }; + +function readTaskSendInput(input: Record): TaskSendInput { + const taskId = + typeof input.taskId === "string" && input.taskId.trim() !== "" ? input.taskId : undefined; + if (taskId === undefined) { + return { kind: "invalid", message: "Provide the `taskId` from a task receipt." }; + } + const message = + typeof input.message === "string" && input.message.trim() !== "" ? input.message : undefined; + const responses = Array.isArray(input.inputResponses) + ? input.inputResponses.filter( + (candidate): candidate is { readonly requestId: string } => + typeof candidate === "object" && + candidate !== null && + typeof (candidate as { requestId?: unknown }).requestId === "string", + ) + : undefined; + if (message !== undefined && responses !== undefined) { + return { kind: "invalid", message: "Provide either `message` or `inputResponses`, not both." }; + } + if (message !== undefined) { + return { body: { kind: "message", message }, kind: "send", taskId }; + } + if (responses !== undefined && responses.length > 0) { + return { body: { inputResponses: responses, kind: "input-responses" }, kind: "send", taskId }; + } + return { kind: "invalid", message: "Provide either `message` or a non-empty `inputResponses`." }; +} diff --git a/packages/eve/src/runtime/framework-tools/tasks.ts b/packages/eve/src/runtime/framework-tools/tasks.ts index d3ab8257f..baf058993 100644 --- a/packages/eve/src/runtime/framework-tools/tasks.ts +++ b/packages/eve/src/runtime/framework-tools/tasks.ts @@ -17,12 +17,14 @@ import type { ResolvedToolDefinition } from "#runtime/types.js"; export const TASK_PEEK_TOOL_NAME = "task_peek"; export const TASK_CANCEL_TOOL_NAME = "task_cancel"; +export const TASK_SEND_TOOL_NAME = "task_send"; export const TASK_SLEEP_TOOL_NAME = "task_sleep"; /** Every model-visible task tool name, for gating and dispatch matching. */ export const TASK_TOOL_NAMES: ReadonlySet = new Set([ TASK_PEEK_TOOL_NAME, TASK_CANCEL_TOOL_NAME, + TASK_SEND_TOOL_NAME, TASK_SLEEP_TOOL_NAME, ]); @@ -30,6 +32,7 @@ export const TASK_TOOL_NAMES: ReadonlySet = new Set([ export const TASK_CONTROL_TOOL_NAMES: ReadonlySet = new Set([ TASK_PEEK_TOOL_NAME, TASK_CANCEL_TOOL_NAME, + TASK_SEND_TOOL_NAME, ]); const TASK_IDS_SCHEMA = z @@ -40,6 +43,28 @@ const TASK_IDS_SCHEMA = z export const TASK_PEEK_INPUT_SCHEMA = z.strictObject({ taskIds: TASK_IDS_SCHEMA }); export const TASK_CANCEL_INPUT_SCHEMA = z.strictObject({ taskIds: TASK_IDS_SCHEMA }); +export const TASK_SEND_INPUT_SCHEMA = z.strictObject({ + inputResponses: z + .array( + z.strictObject({ + optionId: z.string().optional(), + requestId: z.string(), + text: z.string().optional(), + }), + ) + .optional() + .describe( + "Your answers to an input_required task's outstanding requests; each requestId comes from the task's inputRequests. Provide exactly one of inputResponses or message.", + ), + message: z + .string() + .optional() + .describe( + "Follow-up message for a finished task's agent; starts a new task in the same conversation.", + ), + taskId: z.string().min(1).describe("Task id from an earlier task receipt."), +}); + const MAX_SLEEP_SECONDS = Math.floor(Number.MAX_SAFE_INTEGER / 1_000); export const TASK_SLEEP_INPUT_SCHEMA = z.strictObject({ @@ -82,6 +107,12 @@ const TASK_CANCEL_DESCRIPTION = "Request cooperative cancellation of one or more background tasks. " + "Cancellation is final: a task that finishes after you cancel it stays cancelled. Cancelling an already-finished task changes nothing."; +const TASK_SEND_DESCRIPTION = + "Reply to one of your background tasks. " + + "An input_required task means its agent stopped to ask you something: read the questions from the task's inputRequests (via task_peek) and answer them with inputResponses. " + + "A finished task accepts a follow-up message instead, which starts a new task in the same conversation and returns its receipt. " + + "A task that is still working cannot receive sends."; + const TASK_SLEEP_DESCRIPTION = "Pause durably before continuing, for paced background-task checks. " + "Does not read or change any task; follow it with task_peek."; @@ -108,6 +139,12 @@ export function createTaskToolHarnessDefinitions(): readonly HarnessToolDefiniti outputSchema: TASK_VIEWS_OUTPUT_SCHEMA, runtimeAction: { kind: "task-control" }, }, + { + description: TASK_SEND_DESCRIPTION, + inputSchema: TASK_SEND_INPUT_SCHEMA, + name: TASK_SEND_TOOL_NAME, + runtimeAction: { kind: "task-control" }, + }, { description: TASK_SLEEP_DESCRIPTION, execute: async (input: { readonly seconds: number }) => { @@ -166,5 +203,6 @@ function createResolvedTaskToolStub(input: { export const TASK_TOOL_DEFINITIONS: readonly ResolvedToolDefinition[] = [ createResolvedTaskToolStub({ description: TASK_PEEK_DESCRIPTION, name: TASK_PEEK_TOOL_NAME }), createResolvedTaskToolStub({ description: TASK_CANCEL_DESCRIPTION, name: TASK_CANCEL_TOOL_NAME }), + createResolvedTaskToolStub({ description: TASK_SEND_DESCRIPTION, name: TASK_SEND_TOOL_NAME }), createResolvedTaskToolStub({ description: TASK_SLEEP_DESCRIPTION, name: TASK_SLEEP_TOOL_NAME }), ];