Workflow Supervisor PR 5: workflows as agents

A saved workflow can now be run as an agent: `Agent({ graph })` hydrates the
graph (inline, or `{ workflowId }` read under the context's user), runs it on
the kernel through `ExecutionSession` with the agent itself as the run's
`SupervisorHandle`, forwards the run's messages, and returns the run outputs
from `getResults()`. There is no planning phase — the graph is the plan — and
the branch obeys the same `AgentPolicy` as the other three modes: its turn and
token bounds are stamped onto model-less Agent nodes by the same
`applyRunPolicy` the graph-planner branch uses. Supervision is opt-in
(`supervise: true`); without it the run is an ordinary kernel run that never
constructs an escalation.

Websocket: `supervise` on run requests, plus an optional `supervisor` block
(provider, model, bounds, cost cap). `createRunSupervisor` builds the handle for
both server entry points — the websocket runner and the headless/trigger runner
— and returns none unless the flag is explicit and a model resolves.
`supervisor_escalation` / `supervisor_decision` relay to clients unchanged.
Trigger rows carry a `supervise` column, default 0, never migrated on.

Deviations from the plan:

- A flag alone cannot start a supervisor, so the run request also carries
  `SupervisorRunOptions` and the resolution falls back to the connection's
  default model and then to `NODETOOL_SUPERVISOR_PROVIDER` /
  `NODETOOL_SUPERVISOR_MODEL` (trigger runs have no connection defaults). A run
  that asks for supervision it cannot get runs unsupervised rather than failing.
- The supervisor gets a dedicated provider instance and a listener-free context
  copy: per-turn spend is reconciled from the provider's own running cost, and
  the decision's provider traffic is not the run's message stream.
- `ExecutionSessionOptions.supervisor` is PR 4's deliverable and landed here
  because PR 5 needs it; the two branches carry the same three-line change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K83nCW1jVkfQCTKV96iBgP

Author: claude

docs/workflow-supervisor-implementation-plan.md

@@ -91,13 +91,40 @@ Depends on PR 3.
 - Supervisor cost attributed through the existing cost tracking so `nodetool costs` sees it (PRD open question 3 resolved: attributed to the run, tagged `supervisor`).
 - Docs: CLI section in root `CLAUDE.md` + `docs/cli.md`.
 
-### PR 5 — Workflows as agents ∥ (with PR 4)
+### PR 5 — Workflows as agents ∥ (with PR 4) — **shipped**
 
 Depends on PR 3.
 
+Three things the plan did not anticipate, all forced by the websocket half.
+
+A `supervise` flag alone cannot start a supervisor: something has to name the
+model. The run request therefore carries an optional `supervisor`
+(`SupervisorRunOptions`: provider, model, the three bounds, cost cap) next to
+the flag, falling back to the connection's configured default model and then to
+`NODETOOL_SUPERVISOR_PROVIDER` / `NODETOOL_SUPERVISOR_MODEL` — the last exists
+because trigger-driven headless runs have no connection defaults at all. A
+request that asks for supervision it cannot get runs **unsupervised** rather
+than failing, which is the same fail-closed rule every other supervisor failure
+follows.
+
+The trigger flag is a real column (`trigger_registrations.supervise`, default
+`0`, migration `20260801_000001`), read by the dispatcher into the headless run.
+Registration sync mutates existing rows in place, so re-syncing a workflow never
+resets it.
+
+The supervisor gets a **dedicated provider instance** (`getProvider`, not the
+context's cached one) and a listener-free context copy: per-turn spend is
+reconciled from the provider's own running cost, so a second caller on the same
+instance would corrupt the dollar cap, and the decision's own provider traffic
+is not the run's message stream. The escalation and the verdict still cross the
+websocket — they are emitted by the kernel on the run's context.
+
+`ExecutionSessionOptions.supervisor` is PR 4's deliverable and landed here
+because PR 5 needs it; the two branches carry the same three-line change.
+
 - `AgentOptions.graph?: GraphData | { workflowId: string }`; fourth branch in `Agent._executeImpl`: hydrate, run through `ExecutionSession` with self as supervisor, forward messages, `getResults()` returns run outputs. The branch adopts the common `AgentPolicy` object (`packages/agents/src/agent-policy.ts`); it does not add a fifth ad-hoc policy.
 - Websocket: `supervise` flag on run requests; forward `supervisor_*` messages to clients. Trigger rows carry the flag but it **defaults to off** — the flip to default-on belongs to PR 8's gate, nowhere earlier.
-- Tests: `Agent({graph})` returns identical outputs to a bare runner on a clean graph; interventions surface in the message stream.
+- Tests: `Agent({graph})` returns identical outputs to a bare runner on a clean graph; interventions surface in the message stream. Plus what makes the branch trustworthy: a scripted `skip` completes a run that otherwise fails and a scripted `fail` does not, a clean supervised run emits no `supervisor_*` message at all, an aborted signal cancels the run, and `createRunSupervisor` returns no handle without an explicit flag and a resolvable model.
 
 ## Phase C — the product surface
 

package-lock.json

@@ -20,6 +20,7 @@
     "@llamaindex/liteparse": "^1.5.2",
     "@nodetool-ai/app-runtime": "*",
     "@nodetool-ai/config": "*",
+    "@nodetool-ai/execution": "*",
     "@nodetool-ai/kernel": "*",
     "@nodetool-ai/models": "*",
     "@nodetool-ai/node-sdk": "*",

packages/agents/package.json

@@ -26,12 +26,19 @@ import type {
   Chunk
 } from "@nodetool-ai/protocol";
 import { TaskUpdateEvent } from "@nodetool-ai/protocol";
+import { BoundedHandle, type SupervisorBounds } from "@nodetool-ai/kernel";
 import { TaskPlanner } from "./task-planner.js";
+import {
+  resolveAgentGraph,
+  runWorkflowAsAgent,
+  type AgentGraphSource
+} from "./workflow-agent.js";
+import { SupervisorAgent } from "./supervisor/supervisor-agent.js";
 import { TaskExecutor } from "./task-executor.js";
 import { ParallelTaskExecutor } from "./parallel-task-executor.js";
 import { CompilerAgent } from "./compiler-agent.js";
 import { GraphPlanner } from "./graph-planner.js";
-import { AgentWorkflowRunner } from "./agent-workflow-runner.js";
+import { AgentWorkflowRunner, applyRunPolicy } from "./agent-workflow-runner.js";
 import { ScriptPlanner } from "./script-planner.js";
 import { ScriptRunner } from "./script-runner.js";
 import type { Tool } from "./tools/base-tool.js";
@@ -285,6 +292,28 @@ export interface AgentOptions {
    * script API.
    */
   script?: string;
+  /**
+   * Run an existing workflow as this agent: an inline graph, or
+   * `{ workflowId }` to hydrate one from the workflow table. There is no
+   * planning phase — the graph is the plan — and the agent supervises the run
+   * instead of authoring it, so `getResults()` returns the run's outputs.
+   * Requires {@link registry}. Takes precedence over every planning mode.
+   *
+   * Supervision is opt-in: without {@link supervise} the run is an ordinary
+   * kernel run that never constructs an escalation.
+   */
+  graph?: AgentGraphSource;
+  /**
+   * Supervise the {@link graph} run: a failing node invocation escalates to
+   * this agent, which answers with a verdict (retry / substitute / skip /
+   * end_stream / fail). Default `false` — the flip to default-on is gated on
+   * the eval suite, not on this option.
+   */
+  supervise?: boolean;
+  /** Decision, retry, and timeout ceilings for {@link supervise}. */
+  supervisorBounds?: SupervisorBounds;
+  /** Dollar ceiling on supervision for the whole run. */
+  maxSupervisorCostUsd?: number;
   /** Script mode: concurrent `agent()` calls beyond this queue. Default 8. */
   maxConcurrentAgents?: number;
   /** Script mode: lifetime `agent()` call cap per run. Default 100. */
@@ -374,6 +403,10 @@ export class Agent {
   private readonly useGraphPlanner: boolean;
   private readonly useScriptPlanner: boolean;
   private readonly script?: string;
+  private readonly graphSource?: AgentGraphSource;
+  private readonly supervise: boolean;
+  private readonly supervisorBounds?: SupervisorBounds;
+  private readonly maxSupervisorCostUsd?: number;
   private readonly registry?: NodeRegistry;
   private readonly providers?: Record<string, BaseProvider>;
   private readonly securityMonitorEnabled: boolean;
@@ -416,6 +449,10 @@ export class Agent {
     this.useGraphPlanner = opts.useGraphPlanner === true;
     this.useScriptPlanner = opts.useScriptPlanner === true;
     this.script = opts.script;
+    this.graphSource = opts.graph;
+    this.supervise = opts.supervise === true;
+    this.supervisorBounds = opts.supervisorBounds;
+    this.maxSupervisorCostUsd = opts.maxSupervisorCostUsd;
     this.registry = opts.registry;
     this.providers = opts.providers;
     this.securityMonitorEnabled = opts.securityMonitor?.enabled === true;
@@ -687,6 +724,13 @@ export class Agent {
       path.join(os.homedir(), "nodetool_workspace", Date.now().toString());
     await fs.mkdir(workspacePath, { recursive: true });
 
+    // A supplied graph is already the plan: no planner runs, and the agent's
+    // job is to supervise the run rather than author it.
+    if (this.graphSource) {
+      yield* this.executeSuppliedGraph(context, mergedSystemPrompt);
+      return;
+    }
+
     if (this.initialTask) {
       yield* this.executeSingleTask(
         context,
@@ -1263,6 +1307,100 @@ export class Agent {
     }
   }
 
+  /**
+   * Workflows as agents: run a supplied graph on the kernel with this agent as
+   * the run's supervisor. No planning phase — the graph is the plan — so the
+   * only judgment the model supplies is a verdict on a broken invocation.
+   *
+   * The run obeys the same {@link AgentPolicy} as every other mode: the policy's
+   * turn and token bounds are stamped onto model-less Agent nodes exactly as
+   * the graph-planner branch stamps them, and nothing here invents a second set
+   * of numbers. Supervision's own ceilings (decisions, retries, dollars) are the
+   * supervisor's, shared with every other surface that configures one.
+   */
+  private async *executeSuppliedGraph(
+    context: ProcessingContext,
+    systemPrompt: string | undefined
+  ): AsyncGenerator<ProcessingMessage> {
+    if (!this.registry) {
+      throw new Error(
+        "Agent({ graph }) requires a NodeRegistry to resolve node executors."
+      );
+    }
+
+    const graph = applyRunPolicy(
+      await resolveAgentGraph(this.graphSource!, context),
+      {
+        providerId: this.provider.provider,
+        modelId: this.model,
+        ...(systemPrompt ? { systemPrompt } : {}),
+        maxStepIterations: this.policy.maxStepIterations,
+        ...(this.policy.maxTokens !== undefined
+          ? { maxTokens: this.policy.maxTokens }
+          : {})
+      }
+    );
+
+    yield {
+      type: "log_update",
+      node_id: "workflow_executor",
+      node_name: this.name,
+      content: `Running workflow: ${graph.nodes.length} nodes, ${graph.edges.length} edges${
+        this.supervise ? " (supervised)" : ""
+      }...`,
+      severity: "info"
+    } satisfies LogUpdate;
+
+    const supervisor = this.supervise
+      ? new BoundedHandle(
+          new SupervisorAgent({
+            provider: this.provider,
+            model: this.reasoningModel,
+            // The supervisor reads and writes the run's memory (`supervisor:`
+            // keys) but must not push its own provider chatter into the run's
+            // message stream, so it gets a listener-free copy.
+            context: context.copy({
+              shareMemory: true,
+              inheritMessageListeners: false
+            }),
+            ...(this.maxSupervisorCostUsd !== undefined
+              ? { maxCostUsd: this.maxSupervisorCostUsd }
+              : {})
+          }),
+          this.supervisorBounds ?? {}
+        )
+      : undefined;
+
+    const run = runWorkflowAsAgent({
+      graph,
+      registry: this.registry,
+      context,
+      params: this.inputs,
+      ...(supervisor ? { supervisor } : {}),
+      ...(this.signal ? { signal: this.signal } : {})
+    });
+
+    let next = await run.next();
+    while (!next.done) {
+      yield next.value;
+      next = await run.next();
+    }
+    const result = next.value;
+
+    if (result.status === "failed") {
+      throw new Error(result.error ?? "Workflow run failed");
+    }
+
+    this.results = result.outputs ?? {};
+
+    log.info("Agent completed", {
+      name: this.name,
+      status: result.status,
+      interventions: result.interventions?.length ?? 0
+    });
+    this.persistAgentRunMemory();
+  }
+
   /**
    * Execute a single pre-defined task (legacy path for backward compatibility).
    */

packages/agents/src/agent.ts

@@ -684,6 +684,11 @@ export type {
   AgentWorkflowRunnerOptions,
   RunPolicy
 } from "./agent-workflow-runner.js";
+export { resolveAgentGraph, runWorkflowAsAgent } from "./workflow-agent.js";
+export type {
+  AgentGraphSource,
+  WorkflowAgentRunOptions
+} from "./workflow-agent.js";
 export {
   SupervisorAgent,
   DEFAULT_MAX_SUPERVISOR_COST_USD,

packages/agents/src/index.ts

@@ -0,0 +1,131 @@
+/**
+ * Workflows as agents — the run half of `Agent`'s graph branch.
+ *
+ * A saved workflow is already a plan, so this path has no planning phase: the
+ * graph runs on the kernel through `ExecutionSession`, and the agent supplies
+ * judgment only where a node breaks, as the run's `SupervisorHandle`. That is
+ * the whole difference from `AgentWorkflowRunner`, which executes a graph the
+ * planner just wrote and cannot supervise it.
+ *
+ * See docs/workflow-supervisor-design.md §7 entry point 3.
+ */
+
+import { randomUUID } from "node:crypto";
+import { createLogger } from "@nodetool-ai/config";
+import { ExecutionSession, type RawGraphInput } from "@nodetool-ai/execution";
+import type { RunResult, SupervisorHandle } from "@nodetool-ai/kernel";
+import type { NodeRegistry } from "@nodetool-ai/node-sdk";
+import type { ProcessingContext } from "@nodetool-ai/runtime";
+import type { GraphData, ProcessingMessage } from "@nodetool-ai/protocol";
+
+const log = createLogger("nodetool.agents.workflow-agent");
+
+/** Either an inline graph or a saved workflow to hydrate one from. */
+export type AgentGraphSource = GraphData | { workflowId: string };
+
+export interface WorkflowAgentRunOptions {
+  graph: GraphData;
+  /** Resolves every node's executor. Hydration reads flags off it too. */
+  registry: NodeRegistry;
+  /** The run's context. A child copy is made so listeners stay separated. */
+  context: ProcessingContext;
+  /** Start params, keyed by input-node name. */
+  params?: Record<string, unknown>;
+  /**
+   * The run's supervisor, already wrapped in the kernel's `BoundedHandle`.
+   * Omitted, the run behaves exactly as an unsupervised one.
+   */
+  supervisor?: SupervisorHandle;
+  /** External cancellation; aborting it cancels the run. */
+  signal?: AbortSignal;
+}
+
+/**
+ * Hydrate a graph source into a graph. A `{ workflowId }` source is read from
+ * the workflow table under the context's user, so a workflow the caller may
+ * not read stays unreadable here too.
+ */
+export async function resolveAgentGraph(
+  source: AgentGraphSource,
+  context: ProcessingContext
+): Promise<GraphData> {
+  if (!("workflowId" in source)) return source;
+  const { Workflow } = await import("@nodetool-ai/models");
+  const workflow = await Workflow.find(context.userId, source.workflowId);
+  if (!workflow) {
+    throw new Error(`Workflow not found: ${source.workflowId}`);
+  }
+  return workflow.getGraph() as unknown as GraphData;
+}
+
+/**
+ * Run a graph and yield the kernel's messages live; the terminal `RunResult`
+ * is the generator's return value.
+ *
+ * Messages are forwarded to the caller's context as well as yielded, matching
+ * `AgentWorkflowRunner`: a host that only reads the shared context's queue (the
+ * websocket runner, when an Agent node runs inside a workflow) must still see
+ * the inner run.
+ */
+export async function* runWorkflowAsAgent(
+  options: WorkflowAgentRunOptions
+): AsyncGenerator<ProcessingMessage, RunResult> {
+  const { graph, registry, context, supervisor, signal } = options;
+  const jobId = randomUUID();
+
+  // A child context keeps the inner run's listeners off the caller's, the same
+  // separation `AgentWorkflowRunner` makes; memory is shared so sub-agents
+  // inside the graph — and the supervisor's `supervisor:` keys — land in the
+  // run's one memory.
+  const runContext = context.copy({
+    shareMemory: true,
+    inheritMessageListeners: false
+  });
+  const removeListener = runContext.addMessageListener((message) => {
+    context.emit(message);
+  });
+
+  const session = await ExecutionSession.create({
+    graph: graph as unknown as RawGraphInput,
+    registry,
+    jobId,
+    context: runContext,
+    params: options.params ?? {},
+    captureMessages: true,
+    ...(supervisor ? { supervisor } : {})
+  });
+
+  const onAbort = (): void => session.cancel("cancelled");
+  if (signal?.aborted) session.cancel("cancelled");
+  else signal?.addEventListener("abort", onAbort, { once: true });
+
+  log.info("Running workflow as agent", {
+    jobId,
+    nodes: graph.nodes.length,
+    edges: graph.edges.length,
+    supervised: Boolean(supervisor)
+  });
+
+  let drained = false;
+  try {
+    for await (const message of session.messages) {
+      yield message;
+    }
+    drained = true;
+  } finally {
+    signal?.removeEventListener("abort", onAbort);
+    removeListener();
+    // A consumer that stops early (a `break`, or a throw upstream) leaves the
+    // run executing and billing. The stream only closes when the run settles,
+    // so anything short of a full drain means the caller walked away.
+    if (!drained) session.cancel("cancelled");
+  }
+
+  const result = await session.result;
+  log.info("Workflow-as-agent run finished", {
+    jobId,
+    status: result.status,
+    interventions: result.interventions?.length ?? 0
+  });
+  return result;
+}