diff --git a/packages/eve/src/channel/types.ts b/packages/eve/src/channel/types.ts index 3b8abac3f..2e93a6e8d 100644 --- a/packages/eve/src/channel/types.ts +++ b/packages/eve/src/channel/types.ts @@ -8,6 +8,7 @@ import type { InputRequest, InputResponse } from "#runtime/input/types.js"; import type { ChannelAdapter } from "#channel/adapter.js"; import type { AgentLimitsDefinition } from "#shared/agent-definition.js"; import type { JsonObject } from "#shared/json.js"; +import type { TaskView } from "#tasks/types.js"; export type { ContextAccessor } from "#context/key.js"; export type { ChannelInstrumentationProjection } from "#channel/instrumentation.js"; @@ -22,6 +23,8 @@ export type RunSessionLimits = Pick< /** Identifies the session turn to cancel. */ export interface CancelTurnInput { readonly sessionId: string; + /** Framework task whose queued child deliveries should be discarded. */ + readonly taskId?: string; /** Limits the request to the turn the caller observed. */ readonly turnId?: string; } @@ -151,6 +154,8 @@ export type EventEmitFn = (event: UnstampedMessageStreamEvent) => Promise; export interface TurnCaller { readonly callId: string; readonly subagentName: string; + /** Present when this turn is the executor for a durable background task. */ + readonly taskId?: string; readonly replyTo: | { readonly kind: "hook"; readonly token: string } | { readonly kind: "callback"; readonly url: string }; @@ -173,6 +178,18 @@ export interface DeliverPayload { readonly message?: string | UserContent; readonly context?: readonly string[]; readonly outputSchema?: JsonObject; + /** Framework-only task HITL envelopes consumed before adapter/model delivery. */ + readonly taskInputRequests?: readonly { + readonly hookPayload: SubagentInputRequestHookPayload; + readonly taskId: string; + }[]; + /** Framework-only task authorization events projected through the parent channel. */ + readonly taskAuthorizationEvents?: readonly { + readonly hookPayload: SubagentAuthorizationEventHookPayload; + readonly taskId: string; + }[]; + /** Framework-only terminal snapshots cached before task-run retention expires. */ + readonly taskSnapshots?: readonly TaskView[]; readonly [key: string]: unknown; } @@ -184,8 +201,10 @@ export type SessionCommand = readonly kind: "send"; readonly payload: DeliverPayload; readonly requestId?: string; + /** Replay-stable identity for one task-owned child delivery. */ + readonly taskDeliveryId?: string; } - | { readonly kind: "cancel"; readonly turnId?: string } + | { readonly kind: "cancel"; readonly taskId?: string; readonly turnId?: string } | { readonly kind: "compact" } | { readonly kind: "clear" } | { readonly kind: "reset"; readonly reason?: string }; @@ -235,6 +254,7 @@ export interface DeliverHookPayload { readonly caller?: TurnCaller; /** Inbound channel request id used only for workflow attributes. */ readonly requestId?: string; + readonly taskDeliveryId?: string; readonly kind: "deliver"; readonly payloads: readonly DeliverPayload[]; } @@ -339,6 +359,7 @@ export type HookPayload = export interface SessionCallback { readonly callId: string; readonly subagentName: string; + readonly taskId?: string; readonly token: string; readonly url: string; } diff --git a/packages/eve/src/execution/tasks/run-control.ts b/packages/eve/src/execution/tasks/run-control.ts new file mode 100644 index 000000000..041dc0e11 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-control.ts @@ -0,0 +1,215 @@ +import { + EntityConflictError, + HookNotFoundError, + RunExpiredError, + WorkflowRunNotFoundError, +} from "#compiled/@workflow/errors/index.js"; + +import type { TaskRunWorkflowInput } from "#execution/tasks/run-workflow.js"; +import { + startWorkflowPreferLatest, + taskRunWorkflowReference, +} from "#execution/workflow-runtime.js"; +import { getRun, resumeHook } from "#internal/workflow/runtime.js"; +import { walkCauseChain } from "#shared/errors.js"; +import { + TASK_SNAPSHOT_STREAM_NAMESPACE, + isReadyTaskStatus, + type TaskCommand, + type TaskCommandHookPayload, + type TaskRunInboundPayload, + type TaskView, +} from "#tasks/types.js"; + +const TASK_SNAPSHOT_READ_TIMEOUT_MS = 10_000; + +/** + * Node-side controls for durable task runs. Every export must be called + * from inside a `"use step"` body; none of these are steps themselves so + * dispatch and tool steps can compose them inside one durable boundary. + */ + +/** Starts the durable run owning one task's lifecycle. */ +export async function startTaskRun( + input: TaskRunWorkflowInput, +): Promise<{ readonly runId: string }> { + const run = await startWorkflowPreferLatest(taskRunWorkflowReference, [input]); + return { runId: run.runId }; +} + +/** + * Submits one command to a task run. + * + * `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"> { + return (await sendTaskCommandToOwner(input)) === undefined ? "unreachable" : "delivered"; +} + +/** Delivers one command and returns the accepting task workflow's run id. */ +export async function sendTaskCommandToOwner(input: { + readonly command: TaskCommand; + readonly commandToken: string; + readonly retryUnreachable?: { readonly attempts: number; readonly delayMs: number }; +}): Promise<{ readonly runId: string } | undefined> { + const payload: TaskCommandHookPayload = { command: input.command, kind: "task-command" }; + const attempts = Math.max(1, input.retryUnreachable?.attempts ?? 1); + for (let attempt = 0; ; attempt += 1) { + try { + const owner = await resumeHook(input.commandToken, payload); + if ( + typeof owner !== "object" || + owner === null || + !("runId" in owner) || + typeof owner.runId !== "string" + ) { + throw new Error(`Task command hook "${input.commandToken}" returned no owner run id.`); + } + return { runId: owner.runId }; + } catch (error) { + if (!isFinishedTaskRunTarget(error)) { + throw error; + } + if (attempt + 1 >= attempts) { + return undefined; + } + await new Promise((resolve) => setTimeout(resolve, input.retryUnreachable?.delayMs ?? 250)); + } + } +} + +/** + * Hands one non-command inbound payload to a task run. + * + * Used for payloads the run must act on before it may record them — + * today only answered input batches, which it forwards to the child + * first. `unreachable` means the task already finished and disposed its + * hook, so the payload is stale by definition. + */ +export async function sendTaskInboundPayload(input: { + readonly commandToken: string; + readonly payload: TaskRunInboundPayload; +}): Promise<"delivered" | "unreachable"> { + try { + await resumeHook(input.commandToken, input.payload); + return "delivered"; + } catch (error) { + if (!isFinishedTaskRunTarget(error)) { + throw error; + } + return "unreachable"; + } +} + +/** + * Reads the latest snapshot a task run has published, or `undefined` + * when the run has not committed its first snapshot yet (the caller + * already holds the creation receipt, which is `working`). + * + * Snapshots are trusted without re-validation: the task run is the + * single writer and every write passed the transition function. + */ +export async function readLatestTaskSnapshot(input: { + readonly taskRunId: string; +}): Promise { + const stream = getRun(input.taskRunId).getReadable({ + namespace: TASK_SNAPSHOT_STREAM_NAMESPACE, + startIndex: -1, + }); + const tailIndex = await stream.getTailIndex(); + const reader = stream.getReader(); + try { + if (tailIndex < 0) { + return undefined; + } + const result = await readWithTimeout(reader, "latest task snapshot"); + return result; + } finally { + await reader.cancel("eve task snapshot read complete").catch(() => {}); + reader.releaseLock(); + } +} + +/** + * Waits until a task run publishes a ready snapshot — terminal or + * `input_required` — starting from the latest published state. Returns + * immediately when the task is already ready. + * + * Unlike {@link readLatestTaskSnapshot} this read has no timeout; the + * caller owns cancellation by racing this promise (for example against + * turn cancellation) and abandoning it. + */ +export async function waitForReadyTaskSnapshot(input: { + readonly taskRunId: string; +}): Promise { + const stream = getRun(input.taskRunId).getReadable({ + namespace: TASK_SNAPSHOT_STREAM_NAMESPACE, + startIndex: -1, + }); + const reader = stream.getReader(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done || value === undefined) { + throw new Error( + `Task run "${input.taskRunId}" closed its snapshot stream without a ready snapshot.`, + ); + } + if (isReadyTaskStatus(value.status)) { + return value; + } + } + } finally { + await reader.cancel("eve task snapshot wait complete").catch(() => {}); + reader.releaseLock(); + } +} + +async function readWithTimeout( + reader: ReadableStreamDefaultReader, + what: string, +): Promise { + let timeout: ReturnType | undefined; + try { + const result = await Promise.race([ + reader.read().then((read) => ({ kind: "read" as const, read })), + new Promise<{ readonly kind: "timeout" }>((resolve) => { + timeout = setTimeout(() => resolve({ kind: "timeout" }), TASK_SNAPSHOT_READ_TIMEOUT_MS); + }), + ]); + if (result.kind === "timeout") { + throw new Error(`Timed out reading ${what} after ${TASK_SNAPSHOT_READ_TIMEOUT_MS}ms.`); + } + if (result.read.done) { + return undefined; + } + return result.read.value; + } finally { + if (timeout !== undefined) { + clearTimeout(timeout); + } + } +} + +export function isFinishedTaskRunTarget(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-steps.ts b/packages/eve/src/execution/tasks/run-steps.ts new file mode 100644 index 000000000..9649c3d03 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-steps.ts @@ -0,0 +1,216 @@ +import { getWritable } from "#compiled/@workflow/core/index.js"; +import { + EntityConflictError, + HookNotFoundError, + RunExpiredError, + WorkflowRunNotFoundError, +} from "#compiled/@workflow/errors/index.js"; + +import type { + SessionAuthContext, + SessionCommand, + SubagentAuthorizationEventHookPayload, + SubagentInputRequestHookPayload, +} from "#channel/types.js"; +import { resumeHook } from "#internal/workflow/runtime.js"; +import { createLogger } from "#internal/logging.js"; +import { walkCauseChain } from "#shared/errors.js"; +import { + isTerminalTaskStatus, + TASK_SNAPSHOT_STREAM_NAMESPACE, + type TaskInboundAnswerInput, + type TaskInboundAuthorizationEvent, + type TaskInboundInputRequest, + 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 + * the run the single writer readers can trust without re-validating. + */ +export async function appendTaskSnapshotStep(input: { readonly view: TaskView }): Promise { + "use step"; + + const writable = getWritable({ namespace: TASK_SNAPSHOT_STREAM_NAMESPACE }); + const writer = writable.getWriter(); + try { + await writer.write(input.view); + } finally { + writer.releaseLock(); + } +} + +/** Re-emits a task-owned child authorization event through the parent channel. */ +export async function wakeTaskAuthorizationParentStep(input: { + readonly request: TaskInboundAuthorizationEvent; + readonly taskId: string; + readonly token: string; +}): Promise { + "use step"; + + const hookPayload: SubagentAuthorizationEventHookPayload = input.request; + const data = input.request.event.data; + const payload: { + message?: string; + taskAuthorizationEvents: { + hookPayload: SubagentAuthorizationEventHookPayload; + taskId: string; + }[]; + } = { taskAuthorizationEvents: [{ hookPayload, taskId: input.taskId }] }; + if (input.request.event.type === "authorization.required") { + payload.message = `Background task ${input.taskId} needs authorization.`; + } + const command: SessionCommand = { + kind: "send", + payload, + taskDeliveryId: `${input.taskId}:authorization:${input.request.event.type}:${data.turnId}:${data.stepIndex}:${data.sequence}:${data.name}`, + }; + try { + await resumeHook(input.token, command); + } catch (error) { + if (isGoneParentTarget(error)) return; + throw error; + } +} + +/** + * 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: { message: string; taskSnapshots?: readonly TaskView[] } = { + message: formatTaskNotification(input.view), + }; + if (isTerminalTaskStatus(input.view.status)) payload.taskSnapshots = [input.view]; + const command: SessionCommand = { + kind: "send", + payload, + taskDeliveryId: `${input.view.taskId}:ready:${input.view.status}`, + }; + try { + await resumeHook(input.token, command); + } 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; + } +} + +/** Sends an exact local-task HITL batch to the parent's pre-model router. */ +export async function wakeTaskInputRequestParentStep(input: { + readonly request: TaskInboundInputRequest; + readonly taskId: string; + readonly token: string; +}): Promise { + "use step"; + + const command: SessionCommand = { + kind: "send", + payload: { + taskInputRequests: [ + { + hookPayload: input.request as SubagentInputRequestHookPayload, + taskId: input.taskId, + }, + ], + }, + taskDeliveryId: `${input.taskId}:input:${input.request.event.turnId}:${input.request.event.stepIndex}:${input.request.event.sequence}`, + }; + try { + await resumeHook(input.token, command); + } catch (error) { + if (isGoneParentTarget(error)) return; + throw error; + } +} + +/** + * Forwards answered input to the blocked child. + * + * The task run performs this itself so the child unblocks and the + * snapshot leaves `input_required` under one durable decision. Returns + * `unreachable` when the child hook is already gone, which leaves the + * outstanding batch untouched rather than reporting a task as working + * when nothing received the answer. + */ +export async function deliverTaskInputResponsesStep(input: { + readonly answer: TaskInboundAnswerInput; + readonly requestIds: readonly string[]; +}): Promise<"delivered" | "unreachable"> { + "use step"; + + const answered = new Set(input.requestIds); + const command: SessionCommand = { + auth: input.answer.auth as SessionAuthContext | null | undefined, + kind: "send", + payload: { + inputResponses: input.answer.inputResponses.filter((response) => + answered.has(response.requestId), + ), + }, + taskDeliveryId: `${input.answer.taskId}:${[...input.requestIds].sort().join(",")}`, + }; + try { + if (input.answer.childResponseUrl !== undefined) { + const response = await fetch(input.answer.childResponseUrl, { + body: JSON.stringify({ inputResponses: command.payload.inputResponses }), + headers: { "content-type": "application/json" }, + method: "POST", + redirect: "error", + }); + if (response.status === 404) return "unreachable"; + if (!response.ok) + throw new Error(`Remote task input delivery failed with HTTP ${response.status}.`); + } else { + await resumeHook(input.answer.childContinuationToken, command); + } + return "delivered"; + } catch (error) { + if (isGoneParentTarget(error)) { + log.warn("task input answer target is gone; the child turn already ended", { + taskId: input.answer.taskId, + }); + return "unreachable"; + } + 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 new file mode 100644 index 000000000..3c32e6432 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-workflow.test.ts @@ -0,0 +1,453 @@ +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, + deliverTaskInputResponsesStep, + wakeTaskAuthorizationParentStep, + wakeTaskInputRequestParentStep, + wakeTaskParentStep, +} from "#execution/tasks/run-steps.js"; +import { taskRunWorkflow } from "#execution/tasks/run-workflow.js"; +import type { + TaskCommandHookPayload, + TaskInboundAnswerInput, + TaskRunInboundPayload, + TaskView, +} from "#tasks/types.js"; + +vi.mock("#compiled/@workflow/core/index.js", () => ({ + createHook: vi.fn(), +})); + +vi.mock("../hook-ownership.js", async (importOriginal) => ({ + ...(await importOriginal()), + claimHookOwnership: vi.fn(), + disposeHook: vi.fn(), +})); + +vi.mock("./run-steps.js", () => ({ + appendTaskSnapshotStep: vi.fn(), + deliverTaskInputResponsesStep: vi.fn(), + wakeTaskAuthorizationParentStep: vi.fn(), + wakeTaskInputRequestParentStep: vi.fn(), + wakeTaskParentStep: vi.fn(), +})); + +afterEach(() => { + vi.resetAllMocks(); +}); + +function createWorkingView(): TaskView { + return { + metadata: { + agentId: "ag_research:abcdef123456", + kind: "subagent", + mode: "local", + name: "research", + }, + status: "working", + taskId: "task_abc123", + }; +} + +function mockCommandHook(payloads: readonly TaskRunInboundPayload[]): void { + const queue = [...payloads]; + const hook = { + [Symbol.asyncIterator]: () => ({ + next: async () => + queue.length > 0 + ? { done: false as const, value: queue.shift() as TaskCommandHookPayload } + : { done: true as const, value: undefined }, + }), + token: "task-token", + } as Hook; + vi.mocked(createHook).mockReturnValue(hook); +} + +function appendedStatuses(): readonly string[] { + return vi.mocked(appendTaskSnapshotStep).mock.calls.map(([input]) => input.view.status); +} + +describe("taskRunWorkflow", () => { + it("publishes the initial snapshot, applies commands, and stops at terminal", async () => { + mockCommandHook([ + { + command: { + inputRequests: [{ question: "which?", requestId: "req-1" }], + kind: "require-input", + }, + kind: "task-command", + }, + { command: { kind: "answered", requestIds: ["req-1"] }, kind: "task-command" }, + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + // Never consumed: the run stops at the terminal transition. + { command: { kind: "cancel" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(appendedStatuses()).toEqual(["working", "input_required", "working", "completed"]); + expect(disposeHook).toHaveBeenCalledTimes(1); + }); + + it("skips snapshots for rejected and noop commands", async () => { + mockCommandHook([ + { command: { kind: "answered", requestIds: ["req-1"] }, kind: "task-command" }, // noop on working + { command: { kind: "cancel" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(appendedStatuses()).toEqual(["working", "cancelled"]); + }); + + it("exits without touching the lifecycle when the hook claim conflicts", async () => { + mockCommandHook([]); + vi.mocked(claimHookOwnership).mockRejectedValue( + Object.assign(new Error("Hook token in use"), { name: "HookConflictError" }), + ); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(appendTaskSnapshotStep).not.toHaveBeenCalled(); + expect(disposeHook).not.toHaveBeenCalled(); + }); + + it("disposes its hook when the command stream closes early", async () => { + mockCommandHook([ + { command: { inputRequests: [], kind: "require-input" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(appendedStatuses()).toEqual(["working"]); + 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([ + { + 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: { + agentId: "ag_research:abcdef123456", + kind: "subagent", + mode: "local", + name: "research", + }, + }, + wakeToken: "parent-session-token", + }); + + expect(appendedStatuses()).toEqual(["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("keeps a fast terminal task hook alive until dispatch acknowledgement", async () => { + mockCommandHook([ + { + kind: "runtime-action-result", + results: [ + { + outcome: { + kind: "parked", + result: { kind: "succeeded", output: "fast" }, + }, + output: "fast", + }, + ], + }, + { command: { kind: "ready" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ + commandToken: "task-token", + initialView: createWorkingView(), + wakeToken: "parent-session-token", + }); + + expect(appendedStatuses()).toEqual(["working", "completed"]); + expect(disposeHook).toHaveBeenCalledTimes(1); + }); + + it("releases a fast input request when the readiness barrier arrives", async () => { + mockCommandHook([ + { + callId: "call-task", + childContinuationToken: "child-token", + childSessionId: "child-session", + event: { + requests: [ + { + action: { callId: "call-q", input: {}, kind: "tool-call", toolName: "ask" }, + kind: "question", + prompt: "Which?", + requestId: "q1", + }, + ], + sequence: 1, + stepIndex: 2, + turnId: "turn-child", + }, + kind: "subagent-input-request", + subagentName: "research", + }, + { command: { kind: "ready" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ + commandToken: "task-token", + initialView: createWorkingView(), + wakeToken: "parent-session-token", + }); + + expect(wakeTaskInputRequestParentStep).toHaveBeenCalledTimes(1); + expect(wakeTaskParentStep).not.toHaveBeenCalled(); + }); + + it("holds a fast authorization event until the readiness barrier", async () => { + mockCommandHook([ + { + callId: "call-task", + childSessionId: "child-session", + event: { + data: { + description: "Authorize GitHub", + name: "github", + sequence: 1, + stepIndex: 2, + turnId: "turn-child", + }, + type: "authorization.required", + }, + kind: "subagent-authorization-event", + subagentName: "research", + }, + { command: { kind: "ready" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ + commandToken: "task-token", + initialView: createWorkingView(), + wakeToken: "parent-session-token", + }); + + expect(appendedStatuses()).toEqual(["working", "input_required", "input_required"]); + expect(wakeTaskAuthorizationParentStep).toHaveBeenCalledTimes(1); + expect( + vi.mocked(wakeTaskAuthorizationParentStep).mock.invocationCallOrder[0], + ).toBeGreaterThan(vi.mocked(appendTaskSnapshotStep).mock.invocationCallOrder[2] ?? 0); + }); + + it("does not wake without a wake token and never wakes twice for one blocked child", async () => { + mockCommandHook([ + { + command: { inputRequests: [{ q: 1, requestId: "q1" }], kind: "require-input" }, + kind: "task-command", + }, + { + command: { inputRequests: [{ q: 2, requestId: "q2" }], 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, while terminal settlement + // still wakes independently after direct HITL responses. + expect(wakeTaskParentStep).toHaveBeenCalledTimes(2); + + vi.mocked(wakeTaskParentStep).mockClear(); + mockCommandHook([{ command: { data: "done", kind: "complete" }, kind: "task-command" }]); + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + expect(wakeTaskParentStep).not.toHaveBeenCalled(); + }); + + it("commits and forwards every exact local task HITL batch before terminal wake", async () => { + const request = (requestId: string) => ({ + action: { + callId: `call-${requestId}`, + input: {}, + kind: "tool-call" as const, + toolName: "ask", + }, + kind: "question" as const, + prompt: requestId, + requestId, + }); + const inbound = (requestId: string): TaskRunInboundPayload => ({ + callId: "call-task", + childContinuationToken: "child-token", + childSessionId: "child-session-1", + event: { requests: [request(requestId)], sequence: 1, stepIndex: 2, turnId: "turn_child" }, + kind: "subagent-input-request", + subagentName: "research", + }); + mockCommandHook([ + { command: { kind: "ready" }, kind: "task-command" }, + inbound("q1"), + inbound("q2"), + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ + commandToken: "task-token", + initialView: createWorkingView(), + wakeToken: "parent-session-token", + }); + + expect(wakeTaskInputRequestParentStep).toHaveBeenCalledTimes(2); + expect( + vi.mocked(wakeTaskInputRequestParentStep).mock.calls.map(([input]) => { + const request = input.request.event.requests[0]; + return request !== null && typeof request === "object" + ? Reflect.get(request, "requestId") + : undefined; + }), + ).toEqual(["q1", "q2"]); + expect(wakeTaskParentStep).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ view: expect.objectContaining({ status: "completed" }) }), + ); + const firstInputWakeOrder = + vi.mocked(wakeTaskInputRequestParentStep).mock.invocationCallOrder[0] ?? 0; + const firstInputAppendOrder = + vi.mocked(appendTaskSnapshotStep).mock.invocationCallOrder[2] ?? 0; + expect(firstInputAppendOrder).toBeLessThan(firstInputWakeOrder); + }); +}); + +describe("taskRunWorkflow answered input", () => { + function requireInput(...requestIds: readonly string[]): TaskRunInboundPayload { + return { + command: { + inputRequests: requestIds.map((requestId) => ({ prompt: requestId, requestId })), + kind: "require-input", + }, + kind: "task-command", + }; + } + + function answer(...requestIds: readonly string[]): TaskInboundAnswerInput { + return { + childContinuationToken: "child-token", + inputResponses: requestIds.map((requestId) => ({ requestId, text: "answer" })), + kind: "task-answer-input", + taskId: "task_abc123", + }; + } + + it("forwards the answer to the child before recording it as unblocked", async () => { + vi.mocked(deliverTaskInputResponsesStep).mockResolvedValue("delivered"); + mockCommandHook([ + requireInput("q1"), + answer("q1"), + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(deliverTaskInputResponsesStep).toHaveBeenCalledWith({ + answer: answer("q1"), + requestIds: ["q1"], + }); + expect(appendedStatuses()).toEqual(["working", "input_required", "working", "completed"]); + const deliveryOrder = vi.mocked(deliverTaskInputResponsesStep).mock.invocationCallOrder[0] ?? 0; + const unblockOrder = vi.mocked(appendTaskSnapshotStep).mock.invocationCallOrder[2] ?? 0; + expect(deliveryOrder).toBeLessThan(unblockOrder); + }); + + it("never lets an answer to a superseded batch reach the child", async () => { + vi.mocked(deliverTaskInputResponsesStep).mockResolvedValue("delivered"); + mockCommandHook([ + requireInput("q1"), + requireInput("q2"), + answer("q1"), + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(deliverTaskInputResponsesStep).not.toHaveBeenCalled(); + expect(appendedStatuses()).toEqual([ + "working", + "input_required", + "input_required", + "completed", + ]); + }); + + it("stays blocked on the requests an answer did not cover", async () => { + vi.mocked(deliverTaskInputResponsesStep).mockResolvedValue("delivered"); + mockCommandHook([ + requireInput("q1", "q2"), + answer("q1", "unknown"), + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(deliverTaskInputResponsesStep).toHaveBeenCalledWith({ + answer: answer("q1", "unknown"), + requestIds: ["q1"], + }); + const blockedAgain = vi.mocked(appendTaskSnapshotStep).mock.calls[2]?.[0].view; + expect(blockedAgain?.status).toBe("input_required"); + expect(blockedAgain?.inputRequests).toEqual([{ prompt: "q2", requestId: "q2" }]); + }); + + it("keeps the task blocked when the child never received the answer", async () => { + vi.mocked(deliverTaskInputResponsesStep).mockResolvedValue("unreachable"); + mockCommandHook([ + requireInput("q1"), + answer("q1"), + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(appendedStatuses()).toEqual(["working", "input_required", "completed"]); + }); + + it("ignores an answer addressed to a different task", async () => { + vi.mocked(deliverTaskInputResponsesStep).mockResolvedValue("delivered"); + mockCommandHook([ + requireInput("q1"), + { ...answer("q1"), taskId: "task_other" }, + { command: { data: "done", kind: "complete" }, kind: "task-command" }, + ]); + + await taskRunWorkflow({ commandToken: "task-token", initialView: createWorkingView() }); + + expect(deliverTaskInputResponsesStep).not.toHaveBeenCalled(); + expect(appendedStatuses()).toEqual(["working", "input_required", "completed"]); + }); +}); diff --git a/packages/eve/src/execution/tasks/run-workflow.ts b/packages/eve/src/execution/tasks/run-workflow.ts new file mode 100644 index 000000000..4654ae722 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-workflow.ts @@ -0,0 +1,189 @@ +import { createHook } from "#compiled/@workflow/core/index.js"; + +import { claimHookOwnership, disposeHook, isHookConflictError } from "#execution/hook-ownership.js"; +import { + appendTaskSnapshotStep, + deliverTaskInputResponsesStep, + wakeTaskAuthorizationParentStep, + wakeTaskInputRequestParentStep, + wakeTaskParentStep, +} from "#execution/tasks/run-steps.js"; +import { applyTaskTransition } from "#tasks/transitions.js"; +import { translateTaskInboundPayload } from "#tasks/wire.js"; +import { + isReadyTaskStatus, + isTerminalTaskStatus, + readTaskInputRequestId, + readTaskUsage, + type TaskCommand, + type TaskInboundAnswerInput, + type TaskInboundAuthorizationEvent, + type TaskInboundInputRequest, + type TaskRunInboundPayload, + type TaskView, +} from "#tasks/types.js"; + +/** Input for one durable task run. */ +export interface TaskRunWorkflowInput { + /** Private command-hook token; a routing credential, never model-visible. */ + 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 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 + * disposed hook makes any later command fail loudly instead of queueing + * against a finished task. + */ +export async function taskRunWorkflow(input: TaskRunWorkflowInput): Promise { + "use workflow"; + + 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. + const iterator = commands[Symbol.asyncIterator](); + let ownsHook = false; + + try { + try { + await claimHookOwnership(commands); + ownsHook = true; + } catch (error) { + // A duplicate start for the same task (crash between the start + // side effect and its step commit) loses the claim and exits; + // the surviving run owns the lifecycle. + if (isHookConflictError(error)) return; + throw error; + } + + let view = input.initialView; + let pendingInputRequest: TaskInboundInputRequest | undefined; + let pendingAuthorizationEvent: TaskInboundAuthorizationEvent | undefined; + let dispatchAcknowledged = false; + await appendTaskSnapshotStep({ view }); + + while ( + !isTerminalTaskStatus(view.status) || + !dispatchAcknowledged || + (view.status === "cancelled" && view.executor?.lifecycle !== "terminal") + ) { + const next = await iterator.next(); + if (next.done === true) return; + if (next.value.kind === "task-command") dispatchAcknowledged = true; + if (next.value.kind === "subagent-input-request") { + pendingInputRequest = next.value; + } + if (next.value.kind === "subagent-authorization-event") { + pendingAuthorizationEvent = next.value; + } + const command = + view.status === "cancelled" && next.value.kind === "runtime-action-result" + ? { + kind: "settle-executor" as const, + usage: readTaskUsage(next.value.results[0]?.outcome?.usageDelta), + } + : next.value.kind === "task-answer-input" + ? await resolveAnsweredCommand(view, next.value) + : translateTaskInboundPayload(next.value); + if (command === undefined) continue; + const result = applyTaskTransition(view, command); + if (result.outcome !== "accepted") continue; + const becameTerminal = + !isTerminalTaskStatus(view.status) && isTerminalTaskStatus(result.view.status); + const becameReady = !isReadyTaskStatus(view.status) && isReadyTaskStatus(result.view.status); + view = result.view; + await appendTaskSnapshotStep({ view }); + if ( + pendingAuthorizationEvent !== undefined && + dispatchAcknowledged && + input.wakeToken !== undefined + ) { + await wakeTaskAuthorizationParentStep({ + request: pendingAuthorizationEvent, + taskId: view.taskId, + token: input.wakeToken, + }); + pendingAuthorizationEvent = undefined; + } + const routableInputRequest = + pendingInputRequest !== undefined && + dispatchAcknowledged && + view.status === "input_required"; + if (routableInputRequest && input.wakeToken !== undefined) { + await wakeTaskInputRequestParentStep({ + request: pendingInputRequest as TaskInboundInputRequest, + taskId: view.taskId, + token: input.wakeToken, + }); + pendingInputRequest = undefined; + } else if ( + (becameTerminal || + (becameReady && pendingInputRequest === undefined)) && + input.wakeToken !== undefined + ) { + await wakeTaskParentStep({ token: input.wakeToken, view }); + } + // An unrouted request cannot outlive the block it described, or a + // later block would replay it to the parent as if it were new. + if (view.status !== "input_required") pendingInputRequest = undefined; + } + } finally { + // Dispose-only teardown: `iterator.return()` would await a pending + // durable read that never settles, leaving this run `running` + // forever and its hook unswept. + if (ownsHook) await disposeHook(commands); + } +} + +/** + * Forwards one human answer to the blocked child and reports which + * requests it cleared. + * + * Delivery is attempted only for requests the run still lists as + * outstanding, and the resulting command names exactly those ids. That + * ordering is the point: answers to a superseded batch never reach the + * child, and a child whose hook is already gone leaves the task blocked + * instead of moving to `working` with nothing listening. + */ +async function resolveAnsweredCommand( + view: TaskView, + answer: TaskInboundAnswerInput, +): Promise { + if (answer.taskId !== view.taskId || view.status !== "input_required") return undefined; + + const outstanding = new Set( + (view.inputRequests ?? []).flatMap((request) => { + const requestId = readTaskInputRequestId(request); + return requestId === undefined ? [] : [requestId]; + }), + ); + const requestIds = answer.inputResponses + .map((response) => response.requestId) + .filter((requestId) => outstanding.has(requestId)); + if (requestIds.length === 0) return undefined; + + const delivery = await deliverTaskInputResponsesStep({ answer, requestIds }); + return delivery === "delivered" ? { kind: "answered", requestIds } : undefined; +} diff --git a/packages/eve/src/execution/workflow-runtime.ts b/packages/eve/src/execution/workflow-runtime.ts index 392cde255..6c0a36b56 100644 --- a/packages/eve/src/execution/workflow-runtime.ts +++ b/packages/eve/src/execution/workflow-runtime.ts @@ -53,6 +53,7 @@ import type { DynamicSubagentAgentConfig } from "#runtime/subagents/dynamic-agen const WORKFLOW_ENTRY_NAME = "workflowEntry"; const TURN_WORKFLOW_NAME = "turnWorkflow"; const SESSION_TIMEOUT_WORKFLOW_NAME = "sessionTimeoutWorkflow"; +const TASK_RUN_WORKFLOW_NAME = "taskRunWorkflow"; const EVE_PACKAGE_INFO = resolveInstalledPackageInfo(); const COMMAND_HOOK_READY_TIMEOUT_MS = 30_000; @@ -73,6 +74,7 @@ export const STABLE_WORKFLOW_NAMES: ReadonlySet = new Set([ WORKFLOW_ENTRY_NAME, TURN_WORKFLOW_NAME, SESSION_TIMEOUT_WORKFLOW_NAME, + TASK_RUN_WORKFLOW_NAME, ]); const STABLE_ID_BASE = EVE_PACKAGE_INFO.name; @@ -109,6 +111,11 @@ export const sessionTimeoutWorkflowReference = { workflowId: `workflow//${STABLE_ID_BASE}//${SESSION_TIMEOUT_WORKFLOW_NAME}`, }; +/** Stable workflow reference for durable task runs (`experimental.tasks`). */ +export const taskRunWorkflowReference = { + workflowId: `workflow//${STABLE_ID_BASE}//${TASK_RUN_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/protocol/routes.ts b/packages/eve/src/protocol/routes.ts index 36ef897b1..c2ff24005 100644 --- a/packages/eve/src/protocol/routes.ts +++ b/packages/eve/src/protocol/routes.ts @@ -50,6 +50,11 @@ export const EVE_CONTINUE_SESSION_ROUTE_PATTERN = `${EVE_ROUTE_PREFIX}/session/: */ export const EVE_MESSAGE_STREAM_ROUTE_PATTERN = `${EVE_ROUTE_PREFIX}/session/:sessionId/stream`; +/** + * Parent-origin proxy route for one remotely executed child session stream. + */ +export const EVE_SUBAGENT_STREAM_ROUTE_PATTERN = `${EVE_ROUTE_PREFIX}/session/:parentSessionId/subagents/:callId/:childSessionId/stream`; + /** * Stable framework-owned route pattern for cancelling a session's * in-flight turn. Accepts an optional `{ turnId }` body guard scoping @@ -127,6 +132,9 @@ export const EVE_CONNECTION_CALLBACK_ROUTE_PATTERN = `${EVE_ROUTE_PREFIX}/connec */ export const EVE_CALLBACK_ROUTE_PATTERN = `${EVE_ROUTE_PREFIX}/callback/:token`; +/** Capability route used by a parent task to answer a remote child HITL batch. */ +export const EVE_TASK_INPUT_ROUTE_PATTERN = `${EVE_ROUTE_PREFIX}/task-input/:token`; + /** * Creates the stable framework-owned message stream route path for one session. */ @@ -134,6 +142,15 @@ export function createEveMessageStreamRoutePath(sessionId: string): string { return `${EVE_ROUTE_PREFIX}/session/${encodeURIComponent(sessionId)}/stream`; } +/** Builds the parent-origin stream path for one remote child session. */ +export function createEveSubagentStreamRoutePath(input: { + readonly callId: string; + readonly childSessionId: string; + readonly parentSessionId: string; +}): string { + return `${EVE_ROUTE_PREFIX}/session/${encodeURIComponent(input.parentSessionId)}/subagents/${encodeURIComponent(input.callId)}/${encodeURIComponent(input.childSessionId)}/stream`; +} + /** * Creates the stable framework-owned continue-session route path. */ @@ -168,3 +185,8 @@ export function createEveConnectionCallbackRoutePath(name: string, token: string export function createEveCallbackRoutePath(token: string): string { return `${EVE_ROUTE_PREFIX}/callback/${encodeURIComponent(token)}`; } + +/** Builds the capability path used to answer one remote child turn. */ +export function createEveTaskInputRoutePath(token: string): string { + return `${EVE_ROUTE_PREFIX}/task-input/${encodeURIComponent(token)}`; +} diff --git a/packages/eve/src/tasks/json.ts b/packages/eve/src/tasks/json.ts new file mode 100644 index 000000000..b63aec377 --- /dev/null +++ b/packages/eve/src/tasks/json.ts @@ -0,0 +1,37 @@ +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 = { + agentId: view.metadata.agentId, + kind: view.metadata.kind, + mode: view.metadata.mode, + name: view.metadata.name, + }; + + const json: Record = { + metadata, + status: view.status, + taskId: view.taskId, + }; + if (view.lastOutput !== undefined) { + json.lastOutput = { data: view.lastOutput.data, type: view.lastOutput.type }; + } + if (view.inputRequests !== undefined) { + json.inputRequests = [...view.inputRequests]; + } + // `view.executor` and `view.usage` are deliberately not disclosed: they + // are private routing/accounting state, not model-visible task data. + 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/session-index.test.ts b/packages/eve/src/tasks/session-index.test.ts new file mode 100644 index 000000000..e4e456f0c --- /dev/null +++ b/packages/eve/src/tasks/session-index.test.ts @@ -0,0 +1,151 @@ +import { describe, expect, it } from "vitest"; + +import type { HarnessSession } from "#harness/types.js"; +import { + SESSION_TASKS_STATE_KEY, + findSessionTaskEntry, + getSessionTaskIndex, + recordSessionTask, +} from "#tasks/session-index.js"; +import { deriveTaskId } from "#tasks/task-id.js"; +function createSession(state?: HarnessSession["state"]): HarnessSession { + return { + agent: { + modelReference: { id: "model_test" }, + system: "", + tools: [], + }, + compaction: { recentWindowSize: 4, threshold: 1_000_000 }, + continuationToken: "continuation_test", + history: [], + sessionId: "session_parent", + state, + }; +} + +describe("session task index", () => { + const metadata = { + agentId: "ag_research:abcdef123456", + kind: "subagent" as const, + mode: "local" as const, + name: "research", + }; + it("returns an empty index when the key is absent", () => { + expect(getSessionTaskIndex({})).toEqual([]); + expect(getSessionTaskIndex(undefined)).toEqual([]); + }); + + it("records a task and finds it by id", () => { + const session = recordSessionTask(createSession(), { + commandToken: "task:token-1", + createdByTurnId: "turn-1", + metadata, + operationId: "operation-1", + taskId: "task_a", + taskRunId: "run-1", + }); + + expect(findSessionTaskEntry(session.state, "task_a")).toEqual({ + commandToken: "task:token-1", + createdByTurnId: "turn-1", + metadata, + operationId: "operation-1", + taskId: "task_a", + taskRunId: "run-1", + }); + expect(findSessionTaskEntry(session.state, "task_other")).toBeUndefined(); + }); + + it("replaces the entry on replayed creation instead of duplicating it", () => { + let session = recordSessionTask(createSession(), { + commandToken: "task:token-1", + createdByTurnId: "turn-1", + metadata, + operationId: "operation-1", + taskId: "task_a", + taskRunId: "run-1", + }); + session = recordSessionTask(session, { + commandToken: "task:token-2", + createdByTurnId: "turn-1", + metadata, + operationId: "operation-1", + taskId: "task_a", + taskRunId: "run-2", + }); + + const entries = getSessionTaskIndex(session.state); + expect(entries).toHaveLength(1); + expect(entries[0]?.taskRunId).toBe("run-2"); + }); + + it("retains only terminal snapshots as expired-run fallbacks", () => { + const base = { + commandToken: "task:token-1", + createdByTurnId: "turn-1", + metadata, + operationId: "operation-1", + taskId: "task_a", + taskRunId: "run-1", + }; + const terminalSnapshot = { + lastOutput: { data: "done", type: "result" as const }, + metadata, + status: "completed" as const, + taskId: "task_a", + }; + + const session = recordSessionTask(createSession(), { ...base, terminalSnapshot }); + expect(findSessionTaskEntry(session.state, "task_a")?.terminalSnapshot).toEqual( + terminalSnapshot, + ); + for (const invalidSnapshot of [ + { metadata, status: "working", taskId: "task_a" }, + { metadata, status: "completed", taskId: "task_a" }, + { + lastOutput: { data: "wrong", type: "result" }, + metadata, + status: "failed", + taskId: "task_a", + }, + { + lastOutput: { data: "wrong", type: "result" }, + metadata, + status: "cancelled", + taskId: "task_a", + }, + { + inputRequests: [{ requestId: "stale" }], + lastOutput: { data: "done", type: "result" }, + metadata, + status: "completed", + taskId: "task_a", + }, + { ...terminalSnapshot, taskId: "task_other" }, + ]) { + expect(() => + getSessionTaskIndex({ + [SESSION_TASKS_STATE_KEY]: { + tasks: [{ ...base, terminalSnapshot: invalidSnapshot }], + }, + }), + ).toThrow(`Corrupt task index under session state key "${SESSION_TASKS_STATE_KEY}"`); + } + }); + + it("throws on a corrupt index instead of treating it as absent", () => { + expect(() => + getSessionTaskIndex({ [SESSION_TASKS_STATE_KEY]: { tasks: [{ taskId: 42 }] } }), + ).toThrow(`Corrupt task index under session state key "${SESSION_TASKS_STATE_KEY}"`); + }); +}); + +describe("deriveTaskId", () => { + it("is deterministic for the same originating call and distinct otherwise", () => { + const input = { callId: "call-1", parentSessionId: "session-1", parentTurnId: "turn-1" }; + + expect(deriveTaskId(input)).toBe(deriveTaskId(input)); + expect(deriveTaskId(input)).toMatch(/^task_[0-9a-f]{24}$/); + expect(deriveTaskId({ ...input, callId: "call-2" })).not.toBe(deriveTaskId(input)); + }); +}); diff --git a/packages/eve/src/tasks/session-index.ts b/packages/eve/src/tasks/session-index.ts new file mode 100644 index 000000000..6b2c910d0 --- /dev/null +++ b/packages/eve/src/tasks/session-index.ts @@ -0,0 +1,201 @@ +import { z } from "#compiled/zod/index.js"; + +import type { HarnessSession, SessionStateMap } from "#harness/types.js"; +import type { JsonValue } from "#shared/json.js"; +import type { TaskMetadata, TaskView } from "#tasks/types.js"; + +/** + * Session-state key for the parent's live-task index. + * + * The parent session stores only this index; the mutable task record + * lives in the dedicated durable task run. The PR #1190 spike found the + * session-state boundary unworkable for task state itself: session state + * threads through step results, while callback routes and child + * executors must update tasks without holding the current snapshot. + */ +export const SESSION_TASKS_STATE_KEY = "eve.tasks"; + +/** + * One task owned by this session. Immutable model-safe metadata keeps the + * task-to-agent join available before the task run publishes its first view. + * + * `commandToken` is the private routing credential for the task run's + * command hook. It must never render into model context, history, task + * snapshots, or compaction summaries — the model addresses tasks by + * `taskId` only, and lookup verifies ownership through this index. + */ +export interface SessionTaskIndexEntry { + readonly taskId: string; + readonly taskRunId: string; + /** Immutable fallback once the owning workflow run expires. */ + readonly terminalSnapshot?: TaskView; + readonly commandToken: string; + readonly createdByStepIndex?: number; + readonly createdByTurnId: string; + readonly metadata: TaskMetadata; + readonly operationId: string; +} + +const taskMetadataSchema: z.ZodType = z.strictObject({ + agentId: z.string().min(1), + kind: z.literal("subagent"), + mode: z.enum(["local", "remote"]), + name: z.string().min(1), +}); + +const taskViewSchema: z.ZodType = z + .strictObject({ + executor: z + .strictObject({ + childSessionId: z.string().min(1).optional(), + childTurnId: z.string().min(1).optional(), + lifecycle: z.enum(["parked", "terminal"]).optional(), + }) + .optional(), + inputRequests: z.array(z.custom()).optional(), + lastOutput: z + .strictObject({ data: z.custom(), type: z.enum(["result", "error"]) }) + .optional(), + metadata: taskMetadataSchema, + status: z.enum(["working", "input_required", "completed", "failed", "cancelled"]), + taskId: z.string().min(1), + usage: z + .strictObject({ + cacheReadTokens: z.number().nonnegative(), + cacheWriteTokens: z.number().nonnegative(), + inputTokens: z.number().nonnegative(), + outputTokens: z.number().nonnegative(), + }) + .optional(), + }) + .refine((view) => isValidTerminalSnapshot(view), { + message: "Cached task snapshots must satisfy terminal status invariants.", + }); + +const sessionTaskIndexEntrySchema: z.ZodType = z.strictObject({ + commandToken: z.string().min(1), + createdByStepIndex: z.number().int().nonnegative().optional(), + createdByTurnId: z.string().min(1), + metadata: taskMetadataSchema, + operationId: z.string().min(1), + taskId: z.string().min(1), + taskRunId: z.string().min(1), + terminalSnapshot: taskViewSchema.optional(), +}); + +const sessionTaskIndexSchema = z + .strictObject({ + tasks: z.array(sessionTaskIndexEntrySchema), + }) + .refine( + (index) => new Set(index.tasks.map((entry) => entry.taskId)).size === index.tasks.length, + { + message: "Task ids must be unique.", + }, + ) + .refine( + (index) => + index.tasks.every( + (entry) => + entry.terminalSnapshot === undefined || + (entry.terminalSnapshot.taskId === entry.taskId && + sameTaskMetadata(entry.terminalSnapshot.metadata, entry.metadata)), + ), + { message: "Cached terminal snapshots must match their task index entry." }, + ); + +interface SessionTaskIndex { + readonly tasks: readonly SessionTaskIndexEntry[]; +} + +/** + * Reads and validates the task index from session state. + * + * A present but invalid index throws: treating corruption as absence + * would silently orphan every live task's routing credential. + */ +export function getSessionTaskIndex( + state: SessionStateMap | undefined, +): readonly SessionTaskIndexEntry[] { + const raw = state?.[SESSION_TASKS_STATE_KEY]; + if (raw === undefined) { + return []; + } + const parsed = sessionTaskIndexSchema.safeParse(raw); + if (!parsed.success) { + throw new Error( + `Corrupt task index under session state key "${SESSION_TASKS_STATE_KEY}": ${parsed.error.message}`, + ); + } + return parsed.data.tasks; +} + +/** Caches one terminal snapshot beside its task-run address. */ +export function cacheTerminalTaskSnapshot( + state: SessionStateMap | undefined, + snapshot: TaskView, +): SessionStateMap | undefined { + if (!isValidTerminalSnapshot(snapshot)) { + throw new Error(`Cannot cache invalid terminal task "${snapshot.taskId}".`); + } + const entries = getSessionTaskIndex(state); + const index = entries.findIndex((entry) => entry.taskId === snapshot.taskId); + if (index < 0) return state; + if (!sameTaskMetadata(entries[index]!.metadata, snapshot.metadata)) { + throw new Error(`Task snapshot metadata does not match index entry "${snapshot.taskId}".`); + } + const tasks = [...entries]; + tasks[index] = { ...tasks[index]!, terminalSnapshot: snapshot }; + return { ...state, [SESSION_TASKS_STATE_KEY]: { tasks } }; +} + +function isValidTerminalSnapshot(view: TaskView): boolean { + if (view.inputRequests !== undefined) return false; + switch (view.status) { + case "completed": + return view.lastOutput?.type === "result"; + case "failed": + return view.lastOutput?.type === "error"; + case "cancelled": + return view.lastOutput === undefined; + case "working": + case "input_required": + return false; + } +} + +function sameTaskMetadata(left: TaskMetadata, right: TaskMetadata): boolean { + return ( + left.agentId === right.agentId && + left.kind === right.kind && + left.mode === right.mode && + left.name === right.name + ); +} + +/** Finds one owned task; `undefined` enforces parent-session ownership. */ +export function findSessionTaskEntry( + state: SessionStateMap | undefined, + taskId: string, +): SessionTaskIndexEntry | undefined { + return getSessionTaskIndex(state).find((entry) => entry.taskId === taskId); +} + +/** + * Records one task, replacing any entry with the same id so replayed + * creation for the same originating call stays idempotent. + */ +export function recordSessionTask( + session: HarnessSession, + entry: SessionTaskIndexEntry, +): HarnessSession { + const existing = getSessionTaskIndex(session.state); + const tasks = [...existing.filter((candidate) => candidate.taskId !== entry.taskId), entry]; + return { + ...session, + state: { + ...session.state, + [SESSION_TASKS_STATE_KEY]: { tasks } satisfies SessionTaskIndex, + }, + }; +} diff --git a/packages/eve/src/tasks/task-id.ts b/packages/eve/src/tasks/task-id.ts new file mode 100644 index 000000000..daf04dec5 --- /dev/null +++ b/packages/eve/src/tasks/task-id.ts @@ -0,0 +1,50 @@ +import { createHash } from "node:crypto"; + +import { deriveAgentOperationId } from "#harness/handles/operation-id.js"; + +/** + * Derives the stable task id for one originating subagent call. + * + * Reuses the agent-handle operation-id derivation — + * `hash(parentSessionId, parentTurnId, callId)` — so replayed creation + * for the same call yields the same task without new machinery, and a + * task can never be confused with a child session id or continuation + * token. + * + * Lives apart from the pure task modules because the derivation needs + * `node:crypto`, which workflow bodies reject; ids are only ever minted + * inside dispatch steps. + */ +export function deriveTaskId(input: { + readonly callId: string; + readonly parentSessionId: string; + readonly parentTurnId: string; +}): 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)}`; +} + +/** Reads the non-secret task id embedded in a private task command token. */ +export function readTaskIdFromCommandToken(token: string): string | undefined { + const match = /^task:(task_[^:]+):[a-f0-9]{32}$/.exec(token); + return match?.[1]; +} diff --git a/packages/eve/src/tasks/transitions.test.ts b/packages/eve/src/tasks/transitions.test.ts new file mode 100644 index 000000000..7108739f5 --- /dev/null +++ b/packages/eve/src/tasks/transitions.test.ts @@ -0,0 +1,283 @@ +import { describe, expect, it } from "vitest"; + +import { applyTaskTransition } from "#tasks/transitions.js"; +import type { TaskCommand, TaskStatus, TaskView } from "#tasks/types.js"; + +function createView(status: TaskStatus, overrides: Partial = {}): TaskView { + return { + metadata: { + agentId: "ag_research:abcdef123456", + kind: "subagent", + mode: "local", + name: "research", + }, + status, + taskId: "task_abc123", + ...overrides, + }; +} + +const TERMINAL_STATUSES: readonly TaskStatus[] = ["completed", "failed", "cancelled"]; +const ALL_COMMANDS: readonly TaskCommand[] = [ + { data: { answer: 42 }, kind: "complete" }, + { data: { message: "boom" }, kind: "fail" }, + { kind: "cancel" }, + { inputRequests: [{ question: "which?" }], kind: "require-input" }, + { kind: "ready" }, + { kind: "answered", requestIds: ["req-1"] }, +]; + +describe("applyTaskTransition", () => { + it("completes a working task with a result output", () => { + const result = applyTaskTransition(createView("working"), { + data: { answer: 42 }, + kind: "complete", + }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("completed"); + expect(result.view.lastOutput).toEqual({ data: { answer: 42 }, type: "result" }); + }); + + it("retains reported child usage on the terminal snapshot only", () => { + const usage = { cacheReadTokens: 1, cacheWriteTokens: 2, inputTokens: 300, outputTokens: 40 }; + for (const command of [ + { data: "done", kind: "complete", usage }, + { data: "boom", kind: "fail", usage }, + { kind: "cancel", usage }, + ] as const) { + const result = applyTaskTransition(createView("working"), command); + expect(result.outcome).toBe("accepted"); + expect(result.view.usage).toEqual(usage); + } + + const withoutUsage = applyTaskTransition(createView("working"), { + data: "done", + kind: "complete", + }); + expect(withoutUsage.view.usage).toBeUndefined(); + + const blocked = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "which?", requestId: "req-1" }], + kind: "require-input", + }); + expect(blocked.view.usage).toBeUndefined(); + }); + + it("fails a working task and carries the error as its output", () => { + const result = applyTaskTransition(createView("working"), { + data: { message: "boom" }, + kind: "fail", + }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("failed"); + expect(result.view.lastOutput).toEqual({ data: { message: "boom" }, type: "error" }); + }); + + it("moves working to input_required carrying the outstanding batch", () => { + const result = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "which region?", requestId: "req-1" }], + kind: "require-input", + }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("input_required"); + expect(result.view.inputRequests).toEqual([ + { question: "which region?", requestId: "req-1" }, + ]); + }); + + it("rejects empty, unidentified, and duplicate input request batches", () => { + for (const inputRequests of [ + [], + [{ question: "missing id" }], + [ + { question: "first", requestId: "same" }, + { question: "second", requestId: "same" }, + ], + ]) { + const result = applyTaskTransition(createView("working"), { + inputRequests, + kind: "require-input", + }); + expect(result.outcome).toBe("rejected"); + expect(result.view.status).toBe("working"); + } + }); + + it("returns input_required to working once the whole batch is answered", () => { + const blocked = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "which region?", requestId: "req-1" }], + kind: "require-input", + }); + expect(blocked.outcome).toBe("accepted"); + + const result = applyTaskTransition(blocked.view, { kind: "answered", requestIds: ["req-1"] }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("working"); + expect(result.view.inputRequests).toBeUndefined(); + }); + + it("keeps the task blocked on the remainder of a partly answered batch", () => { + const blocked = applyTaskTransition(createView("working"), { + inputRequests: [ + { question: "which region?", requestId: "req-1" }, + { question: "which size?", requestId: "req-2" }, + ], + kind: "require-input", + }); + expect(blocked.outcome).toBe("accepted"); + + const result = applyTaskTransition(blocked.view, { kind: "answered", requestIds: ["req-1"] }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("input_required"); + expect(result.view.inputRequests).toEqual([{ question: "which size?", requestId: "req-2" }]); + }); + + it("ignores an answer to a batch that was already replaced", () => { + const first = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "first", requestId: "req-1" }], + kind: "require-input", + }); + const second = applyTaskTransition(first.view, { + inputRequests: [{ question: "second", requestId: "req-2" }], + kind: "require-input", + }); + expect(second.outcome).toBe("accepted"); + + const stale = applyTaskTransition(second.view, { kind: "answered", requestIds: ["req-1"] }); + + expect(stale.outcome).toBe("noop"); + expect(stale.view.status).toBe("input_required"); + expect(stale.view.inputRequests).toEqual([{ question: "second", requestId: "req-2" }]); + }); + + it("replaces the outstanding batch on repeated require-input", () => { + const first = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "first", requestId: "req-1" }], + kind: "require-input", + }); + expect(first.outcome).toBe("accepted"); + + const second = applyTaskTransition(first.view, { + inputRequests: [{ question: "second", requestId: "req-2" }], + kind: "require-input", + }); + + expect(second.outcome).toBe("accepted"); + expect(second.view.inputRequests).toEqual([{ question: "second", requestId: "req-2" }]); + }); + + it("completes and cancels an input_required task", () => { + const blocked = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "which?", requestId: "req-1" }], + kind: "require-input", + }); + expect(blocked.outcome).toBe("accepted"); + + const completed = applyTaskTransition(blocked.view, { data: "done", kind: "complete" }); + expect(completed.outcome).toBe("accepted"); + expect(completed.view.status).toBe("completed"); + + const cancelled = applyTaskTransition(blocked.view, { kind: "cancel" }); + expect(cancelled.outcome).toBe("accepted"); + expect(cancelled.view.status).toBe("cancelled"); + }); + + it("treats an answer to a working task as a noop", () => { + const result = applyTaskTransition(createView("working"), { + kind: "answered", + requestIds: ["req-1"], + }); + + expect(result.outcome).toBe("noop"); + expect(result.view.status).toBe("working"); + }); + + it("rejects a late completion after cancellation", () => { + const cancelled = applyTaskTransition(createView("working"), { kind: "cancel" }); + expect(cancelled.outcome).toBe("accepted"); + + const late = applyTaskTransition(cancelled.view, { data: "too late", kind: "complete" }); + + expect(late.outcome).toBe("rejected"); + expect(late.view.status).toBe("cancelled"); + expect(late.view.lastOutput).toBeUndefined(); + }); + + it("treats repeated cancellation as an idempotent noop", () => { + const cancelled = applyTaskTransition(createView("working"), { kind: "cancel" }); + expect(cancelled.outcome).toBe("accepted"); + + const again = applyTaskTransition(cancelled.view, { kind: "cancel" }); + + expect(again.outcome).toBe("noop"); + expect(again.view.status).toBe("cancelled"); + }); + + it.each(TERMINAL_STATUSES)("keeps %s final against every non-cancel command", (status) => { + const view = createView(status); + for (const command of ALL_COMMANDS) { + if (command.kind === "cancel" && status === "cancelled") continue; + const result = applyTaskTransition(view, command); + expect(result.outcome).toBe("rejected"); + expect(result.view).toBe(view); + } + }); + + it("rejects cancel on completed and failed tasks", () => { + for (const status of ["completed", "failed"] as const) { + const result = applyTaskTransition(createView(status), { kind: "cancel" }); + expect(result.outcome).toBe("rejected"); + } + }); + + it("is deterministic for replayed commands", () => { + const view = createView("working"); + const command: TaskCommand = { data: { answer: 1 }, kind: "complete" }; + + const first = applyTaskTransition(view, command); + const second = applyTaskTransition(view, command); + + expect(first).toEqual(second); + }); + + it("never rebinds a task turn to a different child session", () => { + const result = applyTaskTransition( + createView("working", { + executor: { childSessionId: "child-session-1" }, + metadata: { + agentId: "ag_research:abcdef123456", + kind: "subagent", + mode: "local", + name: "research", + }, + }), + { + childSessionId: "other-child", + childTurnId: "turn_9", + kind: "start-turn", + taskId: "task_abc123", + }, + ); + + expect(result.outcome).toBe("rejected"); + expect(result.view.executor?.childSessionId).toBe("child-session-1"); + }); + + it("retains late usage on a terminal task without reviving it", () => { + const usage = { cacheReadTokens: 1, cacheWriteTokens: 2, inputTokens: 300, outputTokens: 40 }; + const cancelled = applyTaskTransition(createView("working"), { kind: "cancel" }); + if (cancelled.outcome !== "accepted") throw new Error("Expected cancellation to commit."); + + const result = applyTaskTransition(cancelled.view, { kind: "settle-executor", usage }); + + expect(result).toEqual({ + outcome: "accepted", + view: { ...cancelled.view, executor: { lifecycle: "terminal" }, usage }, + }); + }); +}); diff --git a/packages/eve/src/tasks/transitions.ts b/packages/eve/src/tasks/transitions.ts new file mode 100644 index 000000000..0e4624ee4 --- /dev/null +++ b/packages/eve/src/tasks/transitions.ts @@ -0,0 +1,244 @@ +import type { + TaskCommand, + TaskInputRequest, + TaskOutput, + TaskStatus, + TaskUsage, + TaskView, +} from "#tasks/types.js"; +import { isTerminalTaskStatus, readTaskInputRequestId } from "#tasks/types.js"; + +/** + * Outcome of applying one command to a task snapshot. + * + * - `accepted`: the state changed; the new view must be appended. + * - `noop`: the command is recognized and benign (idempotent cancel, + * stale answer); nothing changed and nothing is appended. + * - `rejected`: the command is invalid for the current status; the + * reason is diagnostic only. + */ +export type TaskTransitionResult = + | { readonly outcome: "accepted"; readonly view: TaskView } + | { readonly outcome: "noop"; readonly view: TaskView } + | { readonly outcome: "rejected"; readonly view: TaskView; readonly reason: string }; + +/** + * Pure transition function for the task lifecycle: + * + * ```text + * working <-> input_required + * | | + * +-----> completed + * +-----> failed + * +-----> cancelled + * ``` + * + * Terminal states are final: a late child result can never revive a + * cancelled task, and repeated cancellation is idempotent. The durable + * task run is the only caller that persists accepted views, which is + * what serializes competing completion, cancellation, and input + * transitions. + */ +/** + * Builds the settled snapshot for one terminal command, carrying the + * child's lifecycle verdict and reported usage when present. Usage is + * retention-only: nothing folds it into parent budgets yet. + */ +function terminalView( + view: TaskView, + command: Extract, + settled: { readonly lastOutput?: TaskOutput; readonly status: TaskStatus }, +): TaskView { + const next: { + executor?: TaskView["executor"]; + lastOutput?: TaskOutput; + metadata: TaskView["metadata"]; + status: TaskStatus; + taskId: string; + usage?: TaskUsage; + } = { + executor: + command.lifecycle === undefined + ? view.executor + : { ...view.executor, lifecycle: command.lifecycle }, + metadata: view.metadata, + status: settled.status, + taskId: view.taskId, + }; + if (settled.lastOutput !== undefined) next.lastOutput = settled.lastOutput; + if (command.usage !== undefined) next.usage = command.usage; + return next; +} + +export function applyTaskTransition(view: TaskView, command: TaskCommand): TaskTransitionResult { + if (isTerminalTaskStatus(view.status)) { + if (command.kind === "settle-executor") { + const executor = { ...view.executor, lifecycle: "terminal" as const }; + if (sameUsage(view.usage, command.usage) && view.executor?.lifecycle === "terminal") { + return { outcome: "noop", view }; + } + const next = { ...view, executor }; + return { + outcome: "accepted", + view: command.usage === undefined ? next : { ...next, usage: command.usage }, + }; + } + if (command.kind === "cancel" && view.status === "cancelled") { + return { outcome: "noop", view }; + } + + return { + outcome: "rejected", + reason: `Task "${view.taskId}" is already ${view.status}; "${command.kind}" cannot change a terminal task.`, + view, + }; + } + + switch (command.kind) { + case "complete": + return { + outcome: "accepted", + view: terminalView(view, command, { + lastOutput: { data: command.data, type: "result" }, + status: "completed", + }), + }; + case "fail": + return { + outcome: "accepted", + view: terminalView(view, command, { + lastOutput: { data: command.data, type: "error" }, + status: "failed", + }), + }; + case "cancel": + return { + outcome: "accepted", + view: terminalView(view, command, { status: "cancelled" }), + }; + case "settle-executor": + return { + outcome: "rejected", + reason: `Task "${view.taskId}" is not terminal; usage settles with its terminal command.`, + view, + }; + case "require-input": + if (!isValidInputRequestBatch(command.inputRequests)) { + return { + outcome: "rejected", + reason: `Task "${view.taskId}" received an invalid input request batch.`, + view, + }; + } + return { + outcome: "accepted", + view: { + inputRequests: command.inputRequests, + executor: view.executor, + metadata: view.metadata, + status: "input_required", + taskId: view.taskId, + }, + }; + case "ready": + // The readiness command is also the barrier that releases a fast, + // pre-acknowledgement HITL batch, so the workflow must observe it. + return { outcome: "accepted", view }; + case "answered": { + if (view.status !== "input_required") { + return { outcome: "noop", view }; + } + + const answered = new Set(command.requestIds); + const outstanding = view.inputRequests ?? []; + const remaining = outstanding.filter((request) => { + const requestId = readTaskInputRequestId(request); + return requestId === undefined || !answered.has(requestId); + }); + // An answer that matches nothing outstanding is stale: the batch + // it was written against was already cleared or replaced. Leaving + // the current batch intact is what stops it from unblocking a + // question the human never saw. + if (remaining.length === outstanding.length) { + return { outcome: "noop", view }; + } + if (remaining.length > 0) { + return { + outcome: "accepted", + view: { + inputRequests: remaining, + executor: view.executor, + metadata: view.metadata, + status: "input_required", + taskId: view.taskId, + }, + }; + } + + return { + outcome: "accepted", + view: { + executor: view.executor, + metadata: view.metadata, + status: "working", + taskId: view.taskId, + }, + }; + } + case "start-turn": { + if (command.taskId !== view.taskId) { + return { + outcome: "rejected", + reason: `Task turn identity "${command.taskId}" does not match "${view.taskId}".`, + view, + }; + } + if ( + view.executor?.childSessionId !== undefined && + view.executor.childSessionId !== command.childSessionId + ) { + return { + outcome: "rejected", + reason: `Task child session "${command.childSessionId}" does not match "${view.executor.childSessionId}".`, + view, + }; + } + if ( + view.executor?.childSessionId === command.childSessionId && + view.executor.childTurnId === command.childTurnId + ) { + return { outcome: "noop", view }; + } + return { + outcome: "accepted", + view: { + ...view, + executor: { + ...view.executor, + childSessionId: command.childSessionId, + childTurnId: command.childTurnId, + }, + }, + }; + } + } +} + +function isValidInputRequestBatch(requests: readonly TaskInputRequest[]): boolean { + if (requests.length === 0) return false; + const ids = requests.map(readTaskInputRequestId); + return ( + ids.every((id): id is string => id !== undefined && id.length > 0) && + new Set(ids).size === ids.length + ); +} + +function sameUsage(left: TaskUsage | undefined, right: TaskUsage | undefined): boolean { + if (left === undefined || right === undefined) return left === right; + return ( + left?.cacheReadTokens === right.cacheReadTokens && + left.cacheWriteTokens === right.cacheWriteTokens && + left.inputTokens === right.inputTokens && + left.outputTokens === right.outputTokens + ); +} diff --git a/packages/eve/src/tasks/types.ts b/packages/eve/src/tasks/types.ts new file mode 100644 index 000000000..ae380a2b2 --- /dev/null +++ b/packages/eve/src/tasks/types.ts @@ -0,0 +1,291 @@ +import type { JsonValue } from "#shared/json.js"; +import type { SubagentAuthorizationEvent } from "#channel/types.js"; + +/** + * Task lifecycle contract for `experimental.tasks`. + * + * A task is one durable unit of delegated work owned by a parent session. + * The durable task run is the single writer for lifecycle transitions + * (see `#execution/tasks/run-workflow.js`); every other path submits + * commands and reads snapshots. This module is dependency-free on + * purpose: it is bundled into workflow bodies, which reject Node.js + * builtins and heavyweight validators. + */ + +/** + * Task lifecycle status. + * + * `completed`, `failed`, and `cancelled` are terminal and final. + * `input_required` is not terminal but is ready for parent action; the + * child must wake its parent rather than deadlock while waiting for input. + */ +export type TaskStatus = "working" | "input_required" | "completed" | "failed" | "cancelled"; + +/** + * Immutable identity of the delegated work behind a task. + * + * `agentId` is parent-controlled and exists before the child acknowledges its + * private address, so the durable task can bind to persistent identity before + * the dispatch side effect runs. + */ +export interface TaskMetadata { + /** Stable model-visible identity of the persistent child session. */ + readonly agentId: string; + readonly kind: "subagent"; + readonly mode: "local" | "remote"; + /** Authored subagent name the parent dispatched. */ + readonly name: string; +} + +/** Private executor state retained for cancellation and address reconciliation. */ +export interface TaskExecutorState { + readonly childSessionId?: string; + readonly childTurnId?: string; + readonly lifecycle?: "parked" | "terminal"; +} + +/** + * Terminal task output. Failure is the state (`failed`); the `error` + * output is its consequence — a `failed` task always carries one. + * This intentionally diverges from MCP, which reserves `failed` for + * protocol-level errors. + */ +export type TaskOutput = + | { readonly type: "result"; readonly data: JsonValue } + | { readonly type: "error"; readonly data: JsonValue }; + +/** + * One outstanding request forwarded from a blocked child. Carried + * opaquely: the task layer routes the batch, the input contract owns + * its shape. + */ +export type TaskInputRequest = JsonValue; + +/** + * Provider token usage a child turn reported. Structural on purpose: it + * mirrors `TokenUsage` (#shared/token-usage.js) without importing its + * zod-backed module into a workflow-bundled file. + */ +export interface TaskUsage { + readonly cacheReadTokens: number; + readonly cacheWriteTokens: number; + readonly inputTokens: number; + readonly outputTokens: number; +} + +/** + * Validates one wire-carried usage value into {@link TaskUsage}. + * Anything malformed is dropped rather than rejected: retention is + * best-effort and must never fail a lifecycle transition. + */ +export function readTaskUsage(value: unknown): TaskUsage | undefined { + if (value === null || typeof value !== "object" || Array.isArray(value)) return undefined; + const cacheReadTokens = readUsageAxis(value, "cacheReadTokens"); + const cacheWriteTokens = readUsageAxis(value, "cacheWriteTokens"); + const inputTokens = readUsageAxis(value, "inputTokens"); + const outputTokens = readUsageAxis(value, "outputTokens"); + if ( + cacheReadTokens === undefined || + cacheWriteTokens === undefined || + inputTokens === undefined || + outputTokens === undefined + ) { + return undefined; + } + return { cacheReadTokens, cacheWriteTokens, inputTokens, outputTokens }; +} + +function readUsageAxis(value: object, key: string): number | undefined { + const field = Reflect.get(value, key); + return typeof field === "number" && Number.isFinite(field) && field >= 0 ? field : undefined; +} + +/** + * One human answer to an outstanding request. Structural on purpose: it + * mirrors the input contract's `InputResponse` without importing its + * zod-backed module into a workflow-bundled file. + */ +export interface TaskInputResponse { + readonly optionId?: string; + readonly requestId: string; + readonly text?: string; +} + +/** + * Request id of the synthetic entry that blocks a task while its child + * waits for authorization. Giving the block an id keeps its release + * bound to the same entry, so completing authorization can never clear + * an unrelated request batch the child raised in the meantime. + */ +export const TASK_AUTHORIZATION_REQUEST_ID = "task:authorization"; + +/** Reads the `requestId` of one opaque outstanding request. */ +export function readTaskInputRequestId(request: TaskInputRequest): string | undefined { + if (request === null || typeof request !== "object" || Array.isArray(request)) return undefined; + const requestId = Reflect.get(request, "requestId"); + return typeof requestId === "string" ? requestId : undefined; +} + +/** + * Full durable task snapshot. The task run appends one per accepted + * command; readers always observe a complete view, never a delta. + * Never contains routing credentials, continuation tokens, or + * authorization capabilities. + */ +export interface TaskView { + readonly taskId: string; + readonly status: TaskStatus; + readonly metadata: TaskMetadata; + /** Private executor state; deliberately excluded from model-visible JSON. */ + readonly executor?: TaskExecutorState; + /** Terminal output; present exactly when `status` is terminal. */ + readonly lastOutput?: TaskOutput; + /** Outstanding requests; present exactly when `status` is `input_required`. */ + readonly inputRequests?: readonly TaskInputRequest[]; + /** + * Provider usage the child reported at settlement; present when the + * terminal command carried it. Retained for later accounting only — + * budgets stay best-effort until a reservation model lands — and + * deliberately excluded from model-visible task views (tasks/json.ts). + */ + readonly usage?: TaskUsage; +} + +/** Commands accepted by the durable task run's transition function. */ +export type TaskCommand = + | { + readonly kind: "complete"; + readonly data: JsonValue; + readonly lifecycle?: "parked" | "terminal"; + readonly usage?: TaskUsage; + } + | { + readonly kind: "fail"; + readonly data: JsonValue; + readonly lifecycle?: "parked" | "terminal"; + readonly usage?: TaskUsage; + } + | { + readonly kind: "cancel"; + readonly lifecycle?: "parked" | "terminal"; + readonly usage?: TaskUsage; + } + /** Retains a late executor settlement without changing task terminal status. */ + | { readonly kind: "settle-executor"; readonly usage?: TaskUsage } + | { readonly kind: "require-input"; readonly inputRequests: readonly TaskInputRequest[] } + | { readonly kind: "ready" } + /** + * Clears the listed requests from the outstanding batch. Bound to + * ids rather than unbound like the former `resume`, so an answer can + * only ever release the batch it was written against. + */ + | { readonly kind: "answered"; readonly requestIds: readonly string[] } + | { + readonly kind: "start-turn"; + readonly childSessionId: string; + readonly childTurnId: string; + readonly taskId: string; + }; + +/** Hook payload envelope commanding a durable task run. */ +export interface TaskCommandHookPayload { + readonly kind: "task-command"; + 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. Retained on the terminal task + * snapshot ({@link TaskView.usage}); folding it into parent + * budgets is deferred until a reservation model lands. + */ + readonly usageDelta?: unknown; + }; + readonly output: JsonValue; + }[]; +} + +export interface TaskInboundInputRequest { + readonly callId: string; + readonly childContinuationToken: string; + readonly childSessionId: string; + readonly kind: "subagent-input-request"; + readonly event: { + readonly requests: readonly TaskInputRequest[]; + readonly sequence: number; + readonly stepIndex: number; + readonly turnId: string; + }; + readonly subagentName: string; +} + +export interface TaskInboundTurnStarted { + readonly childSessionId: string; + readonly childTurnId: string; + readonly kind: "task-child-turn-started"; + readonly taskId: string; +} + +export interface TaskInboundAuthorizationEvent { + readonly callId: string; + readonly childSessionId: string; + readonly kind: "subagent-authorization-event"; + readonly event: SubagentAuthorizationEvent; + readonly subagentName: string; +} + +/** + * Human answers routed to the task run rather than straight to the + * child. The run owns both the delivery and the state change, so the + * batch it clears is exactly the batch it forwarded — the parent can no + * longer unblock the child while the run still believes it is blocked. + */ +export interface TaskInboundAnswerInput { + readonly auth?: unknown; + readonly childContinuationToken: string; + readonly childResponseUrl?: string; + readonly inputResponses: readonly TaskInputResponse[]; + readonly kind: "task-answer-input"; + readonly taskId: string; +} + +/** Everything a task run's command hook may receive. */ +export type TaskRunInboundPayload = + | TaskCommandHookPayload + | TaskInboundChildResult + | TaskInboundInputRequest + | TaskInboundTurnStarted + | TaskInboundAuthorizationEvent + | TaskInboundAnswerInput; + +/** Namespaced run stream carrying `TaskView` snapshots. */ +export const TASK_SNAPSHOT_STREAM_NAMESPACE = "eve.task"; + +/** True when the status can never change again. */ +export function isTerminalTaskStatus(status: TaskStatus): boolean { + return status === "completed" || status === "failed" || status === "cancelled"; +} + +/** True when a transition into this status should wake the parent. */ +export function isReadyTaskStatus(status: TaskStatus): boolean { + return status === "input_required" || isTerminalTaskStatus(status); +} diff --git a/packages/eve/src/tasks/wire.test.ts b/packages/eve/src/tasks/wire.test.ts new file mode 100644 index 000000000..1c4d24a1c --- /dev/null +++ b/packages/eve/src/tasks/wire.test.ts @@ -0,0 +1,198 @@ +import { describe, expect, it } from "vitest"; + +import { TASK_AUTHORIZATION_REQUEST_ID } from "#tasks/types.js"; +import { translateTaskInboundPayload } from "#tasks/wire.js"; + +const ZERO_USAGE = { cacheReadTokens: 0, cacheWriteTokens: 0, inputTokens: 0, outputTokens: 0 }; +const USAGE = { cacheReadTokens: 1, cacheWriteTokens: 2, inputTokens: 300, outputTokens: 40 }; + +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", lifecycle: kind, usage: ZERO_USAGE }); + } + }); + + it("retains nonzero child usage on complete, fail, and cancel commands", () => { + const outcomes = [ + { + expected: { data: "done", kind: "complete" }, + result: { kind: "succeeded", output: "done" }, + }, + { expected: { data: "done", kind: "fail" }, result: { error: "boom", kind: "failed" } }, + { expected: { kind: "cancel" }, result: { kind: "cancelled" } }, + ] as const; + for (const { expected, result } of outcomes) { + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [{ outcome: { kind: "terminal", result, usageDelta: USAGE }, output: "done" }], + }), + ).toEqual({ ...expected, lifecycle: "terminal", usage: USAGE }); + } + }); + + it("drops malformed usage rather than failing the transition command", () => { + for (const usageDelta of [null, "n/a", { inputTokens: -1 }, { inputTokens: 1 }]) { + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [ + { + outcome: { + kind: "terminal", + result: { kind: "succeeded", output: "ok" }, + usageDelta, + }, + output: "ok", + }, + ], + }), + ).toEqual({ data: "ok", kind: "complete", lifecycle: "terminal" }); + } + }); + + 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", + lifecycle: "terminal", + usage: ZERO_USAGE, + }); + + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [ + { + outcome: { kind: "terminal", result: { kind: "cancelled" }, usageDelta: ZERO_USAGE }, + output: null, + }, + ], + }), + ).toEqual({ kind: "cancel", lifecycle: "terminal", usage: ZERO_USAGE }); + }); + + it("ignores results without an explicit lifecycle outcome", () => { + expect( + translateTaskInboundPayload({ + kind: "runtime-action-result", + results: [{ isError: true, output: "broken" }], + }), + ).toBeUndefined(); + expect( + translateTaskInboundPayload({ kind: "runtime-action-result", results: [{ output: "ok" }] }), + ).toBeUndefined(); + }); + + 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({ + callId: "call-1", + childContinuationToken: "child-token", + childSessionId: "child-session", + event: { + requests: [{ prompt: "Which region?" }], + sequence: 0, + stepIndex: 0, + turnId: "turn_0", + }, + kind: "subagent-input-request", + subagentName: "research", + }), + ).toEqual({ inputRequests: [{ prompt: "Which region?" }], kind: "require-input" }); + }); + + it("blocks authorization under a reserved id that only its completion clears", () => { + expect( + translateTaskInboundPayload({ + callId: "call-1", + childSessionId: "child-session", + event: { + data: { + description: "Authorize GitHub", + name: "github", + sequence: 1, + stepIndex: 2, + turnId: "turn-1", + }, + type: "authorization.required", + }, + kind: "subagent-authorization-event", + subagentName: "research", + }), + ).toEqual({ + inputRequests: [{ blockedOn: "authorization", requestId: TASK_AUTHORIZATION_REQUEST_ID }], + kind: "require-input", + }); + expect( + translateTaskInboundPayload({ + callId: "call-1", + childSessionId: "child-session", + event: { + data: { + name: "github", + outcome: "authorized", + sequence: 2, + stepIndex: 2, + turnId: "turn-1", + }, + type: "authorization.completed", + }, + kind: "subagent-authorization-event", + subagentName: "research", + }), + ).toEqual({ kind: "answered", requestIds: [TASK_AUTHORIZATION_REQUEST_ID] }); + }); + + it("leaves answered input to the run, which must deliver before recording it", () => { + expect( + translateTaskInboundPayload({ + childContinuationToken: "child-token", + inputResponses: [{ requestId: "req-1", text: "west" }], + kind: "task-answer-input", + taskId: "task-1", + }), + ).toBeUndefined(); + }); +}); diff --git a/packages/eve/src/tasks/wire.ts b/packages/eve/src/tasks/wire.ts new file mode 100644 index 000000000..431965b4a --- /dev/null +++ b/packages/eve/src/tasks/wire.ts @@ -0,0 +1,87 @@ +import type { TaskCommand, TaskRunInboundPayload, TaskUsage } from "#tasks/types.js"; +import { TASK_AUTHORIZATION_REQUEST_ID, readTaskUsage } 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) under a reserved request id, and + * `authorization.completed` clears exactly that id. Authorization + * payloads never enter the snapshot — only the fact that the child is + * blocked does. + * + * `task-answer-input` is deliberately absent: the run must forward the + * answers to the child before it may record them, so it builds that + * command itself rather than translating one here. + * + * 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) { + // Usage retention: the settled outcome's `usageDelta` survives into + // the terminal command so the snapshot keeps the child's spend. + const usage = readTaskUsage(result.outcome.usageDelta); + switch (result.outcome.result.kind) { + case "succeeded": + return withUsage( + { data: result.output, kind: "complete", lifecycle: result.outcome.kind }, + usage, + ); + case "failed": + return withUsage( + { data: result.output, kind: "fail", lifecycle: result.outcome.kind }, + usage, + ); + case "cancelled": + return withUsage({ kind: "cancel", lifecycle: result.outcome.kind }, usage); + } + } + return undefined; + } + case "subagent-input-request": + return { inputRequests: payload.event.requests, kind: "require-input" }; + case "task-child-turn-started": + return { + childSessionId: payload.childSessionId, + childTurnId: payload.childTurnId, + kind: "start-turn", + taskId: payload.taskId, + }; + case "subagent-authorization-event": + return payload.event.type === "authorization.required" + ? { + inputRequests: [ + { blockedOn: "authorization", requestId: TASK_AUTHORIZATION_REQUEST_ID }, + ], + kind: "require-input", + } + : { kind: "answered", requestIds: [TASK_AUTHORIZATION_REQUEST_ID] }; + default: + return undefined; + } +} + +function withUsage( + command: Extract, + usage: TaskUsage | undefined, +): TaskCommand { + if (usage === undefined) return command; + return { ...command, usage }; +}