feat(eve): add task_send for blocked and finished background tasks

.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_await`, `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.

packages/eve/src/cli/dev/tui/tool-presentation.test.ts

@@ -181,6 +181,7 @@ describe("presentTool", () => {
       task_await: { taskIds: ["task_abc"] },
       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" },

packages/eve/src/cli/dev/tui/tool-presentation.ts

@@ -147,6 +147,13 @@ const BUILTIN_TOOL_COPY: Readonly<Record<string, BuiltinToolCopy>> = {
     singularNoun: "task",
     pluralNoun: "tasks",
   },
+  task_send: {
+    verb: "Send",
+    pastVerb: "Sent",
+    argKey: "taskId",
+    singularNoun: "task",
+    pluralNoun: "tasks",
+  },
   task_sleep: {
     verb: "Pause",
     pastVerb: "Paused",

packages/eve/src/execution/dispatch-runtime-actions-step.ts

@@ -200,8 +200,10 @@ export async function dispatchRuntimeActionsStep(input: {
           action: entry.action,
           bundle,
           parentContinuationToken: input.parentContinuationToken,
+          parentTurnId: batch.event.turnId,
           session: nextSession,
         });
+        nextSession = control.session;
         if (control.result !== undefined) {
           results.push(control.result);
         }

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_await", "task_cancel", "task_sleep"]) {
+    for (const name of ["task_peek", "task_await", "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_await", "task_cancel"]) {
+    for (const name of ["task_peek", "task_await", "task_cancel", "task_send"]) {
       expect(tools.get(name)?.runtimeAction).toEqual({ kind: "task-control" });
       expect(tools.get(name)?.execute).toBeUndefined();
     }

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_await`/`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<TaskView[]> {
+  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<AgentHandle, { phase: "running" | "parked" }> | undefined {
+  if (childSessionId === undefined) return undefined;
+  const handles = getAgentHandleStore(session.state)?.handles ?? [];
+  return handles
+    .filter(
+      (candidate): candidate is Extract<AgentHandle, { phase: "running" | "parked" }> =>
+        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.`,
+  );
+}

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<DelegatedTask> {
+  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<void> {
+  await sendTaskCommand({
+    command: { data: input.error, kind: "fail" },
+    commandToken: input.task.commandToken,
+    retryUnreachable: { attempts: 20, delayMs: 250 },
+  });
+}