feat(eve): add the durable task foundation, inert

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<TaskView | undefined> {
+  const stream = getRun<unknown>(input.taskRunId).getReadable<TaskView>({
+    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<TaskView> {
+  const stream = getRun<unknown>(input.taskRunId).getReadable<TaskView>({
+    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<TaskView>,
+  what: string,
+): Promise<TaskView | undefined> {
+  let timeout: ReturnType<typeof setTimeout> | 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;
+}

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<void> {
+  "use step";
+
+  const writable = getWritable<TaskView>({ namespace: TASK_SNAPSHOT_STREAM_NAMESPACE });
+  const writer = writable.getWriter();
+  try {
+    await writer.write(input.view);
+  } finally {
+    writer.releaseLock();
+  }
+}

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<typeof import("../hook-ownership.js")>()),
+  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<TaskCommandHookPayload>;
+  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);
+  });
+});

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<void> {
+  "use workflow";
+
+  const commands = createHook<TaskCommandHookPayload>({ 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);
+  }
+}

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<string> = 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.