From bea6c0e4c731662756a3661849ad02ba4d15e5e5 Mon Sep 17 00:00:00 2001 From: Rui Conti Date: Sat, 1 Aug 2026 14:51:41 -0400 Subject: [PATCH] feat(eve): add the durable task foundation, inert Adds the packages/eve/src/tasks module cluster for experimental.tasks: the TaskStatus/TaskView/TaskOutput contract, the pure lifecycle transition function (terminal is final, failed carries its error output, working <-> input_required), the eve.tasks session index that holds each task's private command token, and replay-idempotent task ids derived from the agent-handle operation-id derivation. The durable task run is a small stable-named workflow that is the single writer for one task's lifecycle: it consumes commands over a private hook, applies the transition function, and appends a full TaskView snapshot per accepted command to its eve.task run stream. Node-side controls (start, command, tail-read, ready-wait) compose inside dispatch and tool steps. Nothing references these modules yet; no runtime behavior changes. Signed-off-by: Rui Conti --- .../eve/src/execution/tasks/run-control.ts | 164 ++++++++++++++++++ packages/eve/src/execution/tasks/run-steps.ts | 20 +++ .../src/execution/tasks/run-workflow.test.ts | 110 ++++++++++++ .../eve/src/execution/tasks/run-workflow.ts | 69 ++++++++ .../eve/src/execution/workflow-runtime.ts | 7 + packages/eve/src/tasks/session-index.test.ts | 80 +++++++++ packages/eve/src/tasks/session-index.ts | 98 +++++++++++ packages/eve/src/tasks/task-id.ts | 22 +++ packages/eve/src/tasks/transitions.test.ts | 163 +++++++++++++++++ packages/eve/src/tasks/transitions.ts | 108 ++++++++++++ packages/eve/src/tasks/types.ts | 96 ++++++++++ 11 files changed, 937 insertions(+) create mode 100644 packages/eve/src/execution/tasks/run-control.ts create mode 100644 packages/eve/src/execution/tasks/run-steps.ts create mode 100644 packages/eve/src/execution/tasks/run-workflow.test.ts create mode 100644 packages/eve/src/execution/tasks/run-workflow.ts create mode 100644 packages/eve/src/tasks/session-index.test.ts create mode 100644 packages/eve/src/tasks/session-index.ts create mode 100644 packages/eve/src/tasks/task-id.ts create mode 100644 packages/eve/src/tasks/transitions.test.ts create mode 100644 packages/eve/src/tasks/transitions.ts create mode 100644 packages/eve/src/tasks/types.ts 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..d3e00912c --- /dev/null +++ b/packages/eve/src/execution/tasks/run-control.ts @@ -0,0 +1,164 @@ +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 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 run already finished and disposed its hook — + * the task is terminal, and the caller should read the final snapshot + * instead of treating the send as a failure. + */ +export async function sendTaskCommand(input: { + readonly command: TaskCommand; + readonly commandToken: string; +}): Promise<"delivered" | "unreachable"> { + const payload: TaskCommandHookPayload = { command: input.command, kind: "task-command" }; + try { + await resumeHook(input.commandToken, payload); + return "delivered"; + } catch (error) { + if (isFinishedTaskRunTarget(error)) { + return "unreachable"; + } + throw error; + } +} + +/** + * 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); + } + } +} + +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..faa2ebbb6 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-steps.ts @@ -0,0 +1,20 @@ +import { getWritable } from "#compiled/@workflow/core/index.js"; + +import { TASK_SNAPSHOT_STREAM_NAMESPACE, type TaskView } from "#tasks/types.js"; + +/** + * 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(); + } +} 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..f5725eef4 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-workflow.test.ts @@ -0,0 +1,110 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createHook, type Hook } from "#compiled/@workflow/core/index.js"; + +import { claimHookOwnership, disposeHook } from "#execution/hook-ownership.js"; +import { appendTaskSnapshotStep } from "#execution/tasks/run-steps.js"; +import { taskRunWorkflow } from "#execution/tasks/run-workflow.js"; +import type { TaskCommandHookPayload, 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(), +})); + +afterEach(() => { + vi.resetAllMocks(); +}); + +function createWorkingView(): TaskView { + return { + metadata: { + childSessionId: "child-session-1", + kind: "subagent", + mode: "local", + name: "research", + }, + status: "working", + taskId: "task_abc123", + }; +} + +function mockCommandHook(payloads: readonly TaskCommandHookPayload[]): 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?" }], kind: "require-input" }, + kind: "task-command", + }, + { command: { kind: "resume-working" }, 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: "resume-working" }, 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", "input_required"]); + expect(disposeHook).toHaveBeenCalledTimes(1); + }); +}); 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..b8df99788 --- /dev/null +++ b/packages/eve/src/execution/tasks/run-workflow.ts @@ -0,0 +1,69 @@ +import { createHook } from "#compiled/@workflow/core/index.js"; + +import { claimHookOwnership, disposeHook, isHookConflictError } from "#execution/hook-ownership.js"; +import { appendTaskSnapshotStep } from "#execution/tasks/run-steps.js"; +import { applyTaskTransition } from "#tasks/transitions.js"; +import { isTerminalTaskStatus, type TaskCommandHookPayload, 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; +} + +/** + * The durable task run: single writer for one task's lifecycle. + * + * Consumes commands over its private hook, applies the pure transition + * function, and appends a full `TaskView` snapshot per accepted command + * to its `eve.task` run stream. Competing completion, cancellation, and + * input transitions serialize here; rejected commands (for example a + * late child result after `cancelled`) change nothing. + * + * 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; + await appendTaskSnapshotStep({ view }); + + while (!isTerminalTaskStatus(view.status)) { + const next = await iterator.next(); + if (next.done === true) return; + const result = applyTaskTransition(view, next.value.command); + if (result.outcome !== "accepted") continue; + view = result.view; + await appendTaskSnapshotStep({ view }); + } + } 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); + } +} diff --git a/packages/eve/src/execution/workflow-runtime.ts b/packages/eve/src/execution/workflow-runtime.ts index ce9eabcd1..2dae778b4 100644 --- a/packages/eve/src/execution/workflow-runtime.ts +++ b/packages/eve/src/execution/workflow-runtime.ts @@ -56,6 +56,7 @@ import { 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(); export const LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE = @@ -75,6 +76,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; @@ -111,6 +113,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/tasks/session-index.test.ts b/packages/eve/src/tasks/session-index.test.ts new file mode 100644 index 000000000..8af416bb8 --- /dev/null +++ b/packages/eve/src/tasks/session-index.test.ts @@ -0,0 +1,80 @@ +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", () => { + 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", + taskId: "task_a", + taskRunId: "run-1", + }); + + expect(findSessionTaskEntry(session.state, "task_a")).toEqual({ + commandToken: "task:token-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", + taskId: "task_a", + taskRunId: "run-1", + }); + session = recordSessionTask(session, { + commandToken: "task:token-2", + taskId: "task_a", + taskRunId: "run-2", + }); + + const entries = getSessionTaskIndex(session.state); + expect(entries).toHaveLength(1); + expect(entries[0]?.taskRunId).toBe("run-2"); + }); + + 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..8c3908aa7 --- /dev/null +++ b/packages/eve/src/tasks/session-index.ts @@ -0,0 +1,98 @@ +import { z } from "#compiled/zod/index.js"; + +import type { HarnessSession, SessionStateMap } from "#harness/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. + * + * `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; + readonly commandToken: string; +} + +const sessionTaskIndexEntrySchema: z.ZodType = z.strictObject({ + commandToken: z.string().min(1), + taskId: z.string().min(1), + taskRunId: z.string().min(1), +}); + +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.", + }, + ); + +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; +} + +/** 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..23f4c58c2 --- /dev/null +++ b/packages/eve/src/tasks/task-id.ts @@ -0,0 +1,22 @@ +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)}`; +} diff --git a/packages/eve/src/tasks/transitions.test.ts b/packages/eve/src/tasks/transitions.test.ts new file mode 100644 index 000000000..0984d2e14 --- /dev/null +++ b/packages/eve/src/tasks/transitions.test.ts @@ -0,0 +1,163 @@ +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: { + childSessionId: "child-session-1", + 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: "resume-working" }, +]; + +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("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?" }], + kind: "require-input", + }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("input_required"); + expect(result.view.inputRequests).toEqual([{ question: "which region?" }]); + }); + + it("returns input_required to working and clears the batch", () => { + const blocked = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "which region?" }], + kind: "require-input", + }); + expect(blocked.outcome).toBe("accepted"); + + const result = applyTaskTransition(blocked.view, { kind: "resume-working" }); + + expect(result.outcome).toBe("accepted"); + expect(result.view.status).toBe("working"); + expect(result.view.inputRequests).toBeUndefined(); + }); + + it("replaces the outstanding batch on repeated require-input", () => { + const first = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "first" }], + kind: "require-input", + }); + expect(first.outcome).toBe("accepted"); + + const second = applyTaskTransition(first.view, { + inputRequests: [{ question: "second" }], + kind: "require-input", + }); + + expect(second.outcome).toBe("accepted"); + expect(second.view.inputRequests).toEqual([{ question: "second" }]); + }); + + it("completes and cancels an input_required task", () => { + const blocked = applyTaskTransition(createView("working"), { + inputRequests: [{ question: "which?" }], + 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 resume-working on a working task as a noop", () => { + const result = applyTaskTransition(createView("working"), { kind: "resume-working" }); + + 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); + }); +}); diff --git a/packages/eve/src/tasks/transitions.ts b/packages/eve/src/tasks/transitions.ts new file mode 100644 index 000000000..97c5ac016 --- /dev/null +++ b/packages/eve/src/tasks/transitions.ts @@ -0,0 +1,108 @@ +import type { TaskCommand, TaskView } from "#tasks/types.js"; +import { isTerminalTaskStatus } 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, + * redundant resume); 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. + */ +export function applyTaskTransition(view: TaskView, command: TaskCommand): TaskTransitionResult { + if (isTerminalTaskStatus(view.status)) { + 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: { + metadata: view.metadata, + lastOutput: { data: command.data, type: "result" }, + status: "completed", + statusMessage: view.statusMessage, + taskId: view.taskId, + }, + }; + case "fail": + return { + outcome: "accepted", + view: { + metadata: view.metadata, + lastOutput: { data: command.data, type: "error" }, + status: "failed", + statusMessage: view.statusMessage, + taskId: view.taskId, + }, + }; + case "cancel": + return { + outcome: "accepted", + view: { + metadata: view.metadata, + status: "cancelled", + statusMessage: view.statusMessage, + taskId: view.taskId, + }, + }; + case "require-input": + return { + outcome: "accepted", + view: { + inputRequests: command.inputRequests, + metadata: view.metadata, + status: "input_required", + statusMessage: view.statusMessage, + taskId: view.taskId, + }, + }; + case "resume-working": { + if (view.status === "working") { + return { outcome: "noop", view }; + } + + return { + outcome: "accepted", + view: { + metadata: view.metadata, + status: "working", + statusMessage: view.statusMessage, + taskId: view.taskId, + }, + }; + } + } +} diff --git a/packages/eve/src/tasks/types.ts b/packages/eve/src/tasks/types.ts new file mode 100644 index 000000000..b339accdc --- /dev/null +++ b/packages/eve/src/tasks/types.ts @@ -0,0 +1,96 @@ +import type { JsonValue } from "#shared/json.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, so + * `task_await` returns for it — a parent must never deadlock while its + * child waits for input. + */ +export type TaskStatus = "working" | "input_required" | "completed" | "failed" | "cancelled"; + +/** Immutable identity of the delegated work behind a task. */ +export interface TaskMetadata { + readonly kind: "subagent"; + readonly mode: "local" | "remote"; + /** Authored subagent name the parent dispatched. */ + readonly name: string; + /** Child session acknowledged at dispatch. */ + readonly childSessionId: string; + /** Remote children only: the child agent's base URL. */ + readonly url?: string; +} + +/** + * 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; + +/** + * 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; + /** Latest child-reported progress message (unused until the progress flow lands). */ + readonly statusMessage?: string; + readonly metadata: TaskMetadata; + /** Terminal output; present exactly when `status` is terminal. */ + readonly lastOutput?: TaskOutput; + /** Outstanding requests; present exactly when `status` is `input_required`. */ + readonly inputRequests?: readonly TaskInputRequest[]; +} + +/** Commands accepted by the durable task run's transition function. */ +export type TaskCommand = + | { readonly kind: "complete"; readonly data: JsonValue } + | { readonly kind: "fail"; readonly data: JsonValue } + | { readonly kind: "cancel" } + | { readonly kind: "require-input"; readonly inputRequests: readonly TaskInputRequest[] } + | { readonly kind: "resume-working" }; + +/** Hook payload envelope commanding a durable task run. */ +export interface TaskCommandHookPayload { + readonly kind: "task-command"; + readonly command: TaskCommand; +} + +/** 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 `task_await` should stop waiting on this status. */ +export function isReadyTaskStatus(status: TaskStatus): boolean { + return status === "input_required" || isTerminalTaskStatus(status); +}