Workflow Supervisor PR 3: the SupervisorAgent (#4634)

Author: georgi

docs/workflow-supervisor-design.md

@@ -64,6 +64,7 @@ export interface Escalation {
   failureSignature?: string;       // stable categorical code; absent ⇒ stickiness disabled
   candidateOutput?: unknown;       // from RecoverableNodeError — the malformed value, redacted+truncated
   inputs: Record<string, unknown>; // the invocation's input values, redacted+truncated
+  declaredOutputs: Record<string, string>; // the node's declared output types, per slot
   attempt: number;                 // 1-based, per (nodeId, invocationKey)
   spentCostUsd: number;            // provider cost recorded by this invocation
   createdAssets: boolean;
@@ -87,6 +88,13 @@ export type Verdict =
 
 `applyTo: "signature"` is the sticky form: the handle caches the verdict keyed by `(nodeId, failureSignature)` and resolves later escalations that match without waking the agent (PRD scenario 2 — 7 identical failures, 1 LLM call). Keying by signature rather than node keeps one login-required error from silently skipping later, unrelated timeouts on the same node. A signature exists only when the error carries a **stable categorical code** — HTTP status, provider error code, validation path — extracted by a small registry of error-shape recognizers. A plain `Error` gets **no** signature (error class alone is not one; every generic throw would collide), and with no signature stickiness is simply off: each such failure escalates individually. Signatures never derive from message text with embedded values. Only `skip` and `fail` may stick; a sticky `retry` or `substitute` would blindly replay a decision made against different inputs.
 
+`declaredOutputs` is what makes `substitute` authorable at all: a repair has to
+be typed per slot, and the escalation is the only thing the supervisor sees.
+It is also what the agent-side acceptance hook validates against before the
+verdict is allowed to become terminal, so the model learns its repair was the
+wrong shape while its conversation is still open. Type metadata, so nothing to
+redact.
+
 `RecoverableNodeError` (node-sdk) is how a node hands the supervisor the thing that needs repairing: a parser that throws on broken JSON attaches the raw response as `candidateOutput` instead of losing it. **No `candidateOutput`, no `substitute`** — without the broken value in hand, "repair" is fabrication: the model would invent a structurally valid output the runtime validator cannot semantically vet. A plain thrown error offers retry/skip/fail only.
 
 New `ProcessingMessage` variants: `supervisor_escalation` and `supervisor_decision` (escalation + verdict + `decidedBy: "agent" | "sticky" | "bounds" | "default"` + cost). Both flow through the existing `_emit` path.

docs/workflow-supervisor-implementation-plan.md

@@ -55,10 +55,23 @@ Still no LLM. Depends on PR 1.
 
 ## Phase B — the agent and the CLI
 
-### PR 3 — SupervisorAgent
+### PR 3 — SupervisorAgent — **shipped**
 
 Depends on PR 2.
 
+Two additions the plan did not name, both forced by what the agent has to see:
+`Escalation.declaredOutputs` (a repair cannot be typed, or accepted, without
+the node's declared output types) and a kernel-side **`RunStateReader`** handed
+to the handle via `SupervisorHandle.attach()` — the tools are pull-based over
+runner state, and a handle is configured before the runner exists. Output
+recording for `read_node_output` is bounded per node and happens only on a
+supervised run.
+
+Host-side schema validation checks the schema **as the caller wrote it**, not
+the sanitized copy sent to the provider: the sanitizer injects
+`additionalProperties: false` for strict structured-output modes, and enforcing
+that would reject results no author ever forbade.
+
 - `packages/agents/src/supervisor/supervisor-agent.ts`: `SupervisorHandle` via one `StepExecutor` per decision, verdict `outputSchema` (mirroring the escalation's allowed set), serialized queue; the `decide()` signal threads through `StepExecutor` into `provider.generateLoop({signal})` so abort kills the in-flight LLM request.
 - **`TurnBudget` in the runtime provider layer** (design §6): `reserve/commit` object accepted via `generateLoop` options, honored before every model turn by the base loop and by native loop overrides (Claude Agent SDK provider explicitly). Worst-case reservation from the pricing catalog with explicit `supervisorMaxOutputTokens` (default 2048, never unset); no pricing entry ⇒ fail closed. This is a `packages/runtime` change with its own provider-contract test, not a wrapper in the agents package — a wrapper around `generateLoop()` cannot see individual turns.
 - **`StepExecutor` extensions** (design §6): full host-side JSON-schema validation for `finish_step` (enums, nested fields, `additionalProperties`, discriminated unions — today it checks top-level type and required keys only) and an injectable **async acceptance callback** so a `substitute` becomes terminal only after the runtime validator (PR 2, storage resolution included) resolves; failures return as tool errors into the still-open conversation, `MAX_REPAIR_ROUNDS` (3) then `fail`. Both extensions are general `StepExecutor` features with their own tests.

packages/agents/src/index.ts

@@ -684,3 +684,21 @@ export type {
   AgentWorkflowRunnerOptions,
   RunPolicy
 } from "./agent-workflow-runner.js";
+export {
+  SupervisorAgent,
+  DEFAULT_MAX_SUPERVISOR_COST_USD,
+  DEFAULT_SUPERVISOR_MAX_OUTPUT_TOKENS
+} from "./supervisor/supervisor-agent.js";
+export type { SupervisorAgentOptions } from "./supervisor/supervisor-agent.js";
+export { buildVerdictSchema } from "./supervisor/verdict-schema.js";
+export { buildSupervisorPrompt } from "./supervisor/prompt.js";
+export {
+  createSupervisorTools,
+  GetRunStateTool,
+  ReadNodeOutputTool
+} from "./supervisor/tools.js";
+export {
+  validateAgainstSchema,
+  formatViolations
+} from "./utils/json-schema-validate.js";
+export type { SchemaViolation } from "./utils/json-schema-validate.js";

packages/agents/src/step-executor.ts

@@ -19,6 +19,7 @@ import type {
   ToolCall,
   ProviderStreamItem
 } from "@nodetool-ai/runtime";
+import type { TurnBudget } from "@nodetool-ai/runtime";
 import { memoryKeys, withAgentSpanGen } from "@nodetool-ai/runtime";
 import { linkAbort } from "./utils/link-abort.js";
 import { createLogger } from "@nodetool-ai/config";
@@ -40,6 +41,10 @@ import { ControlNodeTool } from "./tools/control-tool.js";
 import { FinishStepTool } from "./tools/finish-step-tool.js";
 import { getMemoryTools } from "./tools/memory-tools.js";
 import { truncateToolResult } from "./constants.js";
+import {
+  formatViolations,
+  validateAgainstSchema
+} from "./utils/json-schema-validate.js";
 
 const log = createLogger("nodetool.agents.step-executor");
 
@@ -242,6 +247,19 @@ function validateAndSanitizeSchema(
   return cleanSchemaRecursive(result) as Record<string, unknown>;
 }
 
+/**
+ * The authored schema, deep-copied and given a `type` when it omits one, so
+ * host validation reads exactly what the caller declared and nothing more.
+ */
+function normalizeDeclaredSchema(
+  raw: unknown
+): Record<string, unknown> | null {
+  if (raw === null || typeof raw !== "object" || Array.isArray(raw)) return null;
+  const copy = JSON.parse(JSON.stringify(raw)) as Record<string, unknown>;
+  if (!("type" in copy) && "properties" in copy) copy["type"] = "object";
+  return copy;
+}
+
 // ---------------------------------------------------------------------------
 // Think-tag removal
 // ---------------------------------------------------------------------------
@@ -334,8 +352,35 @@ export interface StepExecutorOptions {
   upstreamMemoryKeys?: string[];
   /** External cancellation. Aborts the provider loop mid-flight. */
   signal?: AbortSignal;
+  /**
+   * Final say on a schema-valid result, for contracts whose acceptance can
+   * only be decided asynchronously — a repaired value that must resolve
+   * against run storage, say. A rejection is not a failure: it returns to the
+   * model as a tool error inside the still-open conversation, so the next
+   * attempt has the reason. After {@link MAX_REPAIR_ROUNDS} rejections the
+   * step fails rather than looping on an answer the model cannot produce.
+   */
+  acceptResult?: (result: unknown) => Promise<StepResultAcceptance>;
+  /**
+   * Spend admission consulted before every provider turn. Forwarded to
+   * `generateLoop`; a refusal ends the loop, which surfaces here as a step
+   * that never completed.
+   */
+  turnBudget?: TurnBudget;
 }
 
+/** Answer from {@link StepExecutorOptions.acceptResult}. */
+export type StepResultAcceptance =
+  | { accepted: true }
+  | { accepted: false; reason: string };
+
+/**
+ * Rejections by the acceptance callback before the step gives up. Three is
+ * enough for the model to read the reason and correct course; more is a model
+ * that cannot satisfy the contract, and looping costs real money.
+ */
+export const MAX_REPAIR_ROUNDS = 3;
+
 export class StepExecutor {
   private history: Message[] = [];
   private step: Step;
@@ -352,6 +397,8 @@ export class StepExecutor {
   private result: unknown = null;
   private finishStepTool: FinishStepTool | null = null;
   private resultSchema: Record<string, unknown> | null = null;
+  /** The schema as the caller wrote it — what host validation checks. */
+  private declaredSchema: Record<string, unknown> | null = null;
   private iterations = 0;
   private generationFailures = 0;
   private sources: string[] = [];
@@ -362,6 +409,11 @@ export class StepExecutor {
   }> = [];
   private threadId?: string;
   private upstreamMemoryKeys: string[];
+  private acceptResult?: (result: unknown) => Promise<StepResultAcceptance>;
+  private turnBudget?: TurnBudget;
+  private repairRounds = 0;
+  /** Set when acceptance rejected the result once too often. */
+  private acceptanceError: string | null = null;
 
   constructor(opts: StepExecutorOptions) {
     this.task = opts.task;
@@ -376,6 +428,8 @@ export class StepExecutor {
     this.threadId = opts.threadId;
     this.upstreamMemoryKeys = opts.upstreamMemoryKeys ?? [];
     this.signal = opts.signal;
+    this.acceptResult = opts.acceptResult;
+    this.turnBudget = opts.turnBudget;
 
     this.resultSchema = this.loadResultSchema();
 
@@ -410,9 +464,17 @@ export class StepExecutor {
         typeof this.step.outputSchema === "string"
           ? JSON.parse(this.step.outputSchema)
           : this.step.outputSchema;
+      // Two schemas, on purpose. The sanitized one is what the provider is
+      // shown — it injects `additionalProperties: false` because strict
+      // structured-output modes demand it. The authored one is the contract,
+      // and it is what the host validates against: an author who wrote
+      // `properties: {}` meant "an object", not "an empty object", and
+      // enforcing a transport artifact would reject a result nobody forbade.
+      this.declaredSchema = normalizeDeclaredSchema(raw);
       return validateAndSanitizeSchema(raw, defaultDescription);
     } catch {
       // Fallback: permissive object schema
+      this.declaredSchema = null;
       return { type: "object", description: defaultDescription };
     }
   }
@@ -462,41 +524,15 @@ export class StepExecutor {
       return [true, null, normalized];
     }
 
-    // Basic structural validation: check required keys from schema
-    if (
-      this.resultSchema["type"] === "object" &&
-      typeof normalized === "object" &&
-      normalized !== null
-    ) {
-      const requiredKeys = this.resultSchema["required"];
-      if (Array.isArray(requiredKeys)) {
-        const obj = normalized as Record<string, unknown>;
-        for (const key of requiredKeys) {
-          // `requiredKeys` is LLM/planner-authored; use Object.hasOwn so a
-          // required key like "toString"/"constructor" can't be satisfied by
-          // the prototype chain when the model never produced it.
-          if (!Object.hasOwn(obj, key as string)) {
-            return [false, `Missing required key: ${key}`, normalized];
-          }
-        }
-      }
-    }
-
-    // Type check
-    const expectedType = this.resultSchema["type"];
-    if (expectedType === "string" && typeof normalized !== "string") {
-      return [false, `Expected string, got ${typeof normalized}`, normalized];
-    }
-    if (
-      expectedType === "object" &&
-      (typeof normalized !== "object" ||
-        normalized === null ||
-        Array.isArray(normalized))
-    ) {
-      return [false, `Expected object, got ${typeof normalized}`, normalized];
-    }
-    if (expectedType === "array" && !Array.isArray(normalized)) {
-      return [false, `Expected array, got ${typeof normalized}`, normalized];
+    // The whole declared schema, not its outline. Backends vary in how much of
+    // a tool-parameter schema they enforce — enums and unions are routinely
+    // dropped — so a contract holds only where the host checks it.
+    const violations = validateAgainstSchema(
+      normalized,
+      this.declaredSchema ?? this.resultSchema
+    );
+    if (violations.length > 0) {
+      return [false, formatViolations(violations), normalized];
     }
 
     return [true, null, normalized];
@@ -939,6 +975,24 @@ export class StepExecutor {
         normalizedResult !== null &&
         normalizedResult !== undefined
       ) {
+        if (this.acceptResult) {
+          const acceptance = await this.acceptResult(normalizedResult);
+          if (!acceptance.accepted) {
+            this.repairRounds++;
+            if (this.repairRounds >= MAX_REPAIR_ROUNDS) {
+              // Out of rounds. Stop the loop rather than pay for another
+              // attempt at a contract this model is not meeting.
+              this.acceptanceError = acceptance.reason;
+              abort.abort();
+              return JSON.stringify({
+                error: `Result rejected: ${acceptance.reason}`
+              });
+            }
+            return JSON.stringify({
+              error: `Result rejected: ${acceptance.reason}. Call finish_step again with a corrected result.`
+            });
+          }
+        }
         emitCompletion(normalizedResult);
         abort.abort();
         return '{"status": "completed"}';
@@ -1031,6 +1085,7 @@ export class StepExecutor {
         maxIterations: this.maxIterations,
         maxTokens: this.maxTokens,
         sequentialTools: true,
+        turnBudget: this.turnBudget,
         signal: abort.signal
       });
 
@@ -1104,9 +1159,11 @@ export class StepExecutor {
     if (!this.step.completed) {
       this.step.endTime = Date.now();
 
-      const message = generationError
-        ? `Step failed: ${generationError.message}`
-        : `Step failed: exceeded ${this.maxIterations} iterations without completion`;
+      const message = this.acceptanceError
+        ? `Step failed: result rejected after ${MAX_REPAIR_ROUNDS} attempts — ${this.acceptanceError}`
+        : generationError
+          ? `Step failed: ${generationError.message}`
+          : `Step failed: exceeded ${this.maxIterations} iterations without completion`;
       this.step.failed = true;
       this.step.error = message;
 

packages/agents/src/supervisor/prompt.ts

@@ -0,0 +1,67 @@
+/**
+ * The supervisor's system prompt.
+ *
+ * A preamble over the standard execution contract, per the `StepExecutor`
+ * rule — it says what the verbs mean and what the job is, not how to finish a
+ * step.
+ */
+
+import type { Escalation, VerdictAction } from "@nodetool-ai/protocol";
+
+const VERB_SEMANTICS: Record<VerdictAction, string> = {
+  retry: "Run this invocation again, unchanged. Only useful when the failure was transient — a timeout, a rate limit, a 5xx. Offered only when the invocation spent nothing and wrote nothing, so it is never a way to pay twice.",
+  substitute:
+    "Replace the node's output with a corrected value you supply. For a node that produced something almost right — malformed JSON, a near-miss shape. The value is type-checked against the node's declared outputs before it enters the graph; a repair that fails the check is not a repair.",
+  skip: "Retire this invocation. Downstream nodes see it as producing nothing — an item drops out of the batch. Correct when one item is genuinely unprocessable and the rest of the run is still worth having.",
+  end_stream:
+    "End this streaming node's output where it stopped. What it already emitted stands; nothing more comes. The only recovery available once a node has emitted.",
+  fail: "Let the run fail, as it would without a supervisor. The right answer whenever recovery would be a guess."
+};
+
+export const SUPERVISOR_PROMPT_PREAMBLE = `# Role
+You supervise a running workflow. A node invocation has failed. Decide what happens to it.
+
+# What you are deciding
+One invocation, not the run. Everything else in the graph is still executing. Your verdict applies to this invocation alone.
+
+# Duties
+- Diagnose before you decide. \`get_run_state\` shows how the run is going; \`read_node_output\` shows what a node upstream of this failure produced. Read them when the error alone does not tell you what happened.
+- Give a one-line rationale with every verdict. It goes in the run report, and it is what a person reads later to understand why an item is missing.
+- Distrust \`skip\`. Skipping is data loss, and it is silent — the run reports success with a smaller result. Use it when an item is genuinely unprocessable, not when a failure is merely inconvenient. If you skip, name the item in your rationale.
+- \`fail\` is not a defeat. A guessed repair that passes validation and is wrong is worse than a failed run: the run reports success and the wrong answer flows downstream.
+
+# What the verbs mean`;
+
+/** The prompt for one decision: verbs the escalation actually allows, then it. */
+export function buildSupervisorPrompt(escalation: Escalation): string {
+  const verbs = escalation.allowedActions
+    .map((action) => `- \`${action}\` — ${VERB_SEMANTICS[action]}`)
+    .join("\n");
+  const omitted = (Object.keys(VERB_SEMANTICS) as VerdictAction[]).filter(
+    (action) => !escalation.allowedActions.includes(action)
+  );
+  const unavailable =
+    omitted.length > 0
+      ? `\n\nNot available for this failure: ${omitted.map((a) => `\`${a}\``).join(", ")}. The kernel computed that from what this node is and what this invocation already did; asking for one of them gets you \`fail\`.`
+      : "";
+
+  return `${SUPERVISOR_PROMPT_PREAMBLE}
+${verbs}${unavailable}
+
+# The failure
+Node \`${escalation.nodeId}\` (\`${escalation.nodeType}\`), attempt ${escalation.attempt}${
+    escalation.invocationKey ? `, invocation \`${escalation.invocationKey}\`` : ""
+  }.
+
+Error: ${escalation.detail}
+
+Inputs: ${JSON.stringify(escalation.inputs)}${
+    escalation.candidateOutput === undefined
+      ? ""
+      : `\n\nThe node produced this, and it was rejected: ${JSON.stringify(escalation.candidateOutput)}`
+  }${
+    escalation.emitted
+      ? "\n\nThis node already emitted output downstream. Nothing can undo that."
+      : ""
+  }`;
+}