Interactive escalations for the agent-facing run/debug tools (#4639)

Author: georgi

CLAUDE.md

@@ -352,7 +352,20 @@ tool. It posts to `POST /api/workflows/:id/debug`, which runs the workflow and
 returns the same execution summary and verdict the CLI harness computes —
 per-node status and errors, logs, LLM calls, outputs — plus the job record and
 the graph overview. The summary reducer and triage live in
-`@nodetool-ai/execution/debug`, so CLI and agent surfaces cannot drift. The browser surface is exposed in `web/` as
+`@nodetool-ai/execution/debug`, so CLI and agent surfaces cannot drift.
+
+With `interactive: true`, `run_workflow` and `debug_workflow` put the calling
+agent on the failure path the way `--supervise` puts an LLM supervisor there:
+a failing node invocation parks the run and the tool returns the escalation
+(`status: "escalated"` with the supervisor's `Escalation` record — redacted
+inputs, error detail, `allowedActions`). The agent answers via
+**`resolve_workflow_escalation`** — retry, substitute, skip, end_stream, or
+fail, kernel-enforced against the allowed set — and gets back either the next
+escalation or the run's final report. HTTP surface:
+`POST /api/workflows/:id/run|debug {interactive: true}` plus
+`GET/POST /api/debug/sessions/:id[/verdict|/cancel]`
+(`packages/websocket/src/debug-sessions.ts`). Escalations the agent leaves
+unanswered fail closed on the decision timeout (default 10 min). The browser surface is exposed in `web/` as
 `npm run test:debug-harness` (env: `NODETOOL_DEBUG_GRAPH`, `NODETOOL_DEBUG_OUT`,
 `NODETOOL_DEBUG_PARAMS`).
 

docs/workflow-supervisor-design.md

@@ -239,6 +239,7 @@ Surfaces do not construct `WorkflowRunner` directly anymore — CLI, debug, head
 2. **CLI** — `nodetool run --supervise [--max-decisions N] [--max-retries N] [--supervisor-cost-cap USD] [--supervisor-model id]` and the same flags on `nodetool debug`. Interventions print inline (`⛨` lines) and appear in `--json` reports.
 3. **`Agent({ graph })`** — a fourth branch in `Agent._executeImpl` alongside `executeScriptPlan`/`executeGraphPlan`: hydrate the graph, start a `WorkflowRunner` with itself as supervisor, forward the runner's message stream, return run outputs from `getResults()`. No planning phase — the graph is the plan.
 4. **API/web** — `supervise: true` on the run request; the websocket runner constructs the handle and forwards `supervisor_*` messages to the client for the intervention feed. Trigger rows carry the flag, **off by default** until the plan's PR 8 gate passes; the eventual flip covers newly created triggers only (§6.1 consent).
+5. **Interactive (agent tools)** — `interactive: true` on `POST /api/workflows/:id/run|debug` makes the *calling* agent the supervisor: `InteractiveEscalationHandle` (`packages/websocket/src/debug-sessions.ts`) parks each `decide()` until a verdict arrives on `POST /api/debug/sessions/:id/verdict`, and the escalated HTTP response carries the `Escalation` record itself. The agent-facing `run_workflow`/`debug_workflow` tools expose the flag and `resolve_workflow_escalation` answers; the handle is wrapped in the same `BoundedHandle` (caps, sticky verdicts, a timeout sized for a tool round trip — 10 min — that fails closed), and the kernel still enforces `allowedActions` whatever arrives.
 
 ## 8. Replay
 

packages/agents/src/index.ts

@@ -108,6 +108,7 @@ export {
   CreateWorkflowTool,
   RunWorkflowTool,
   DebugWorkflowTool,
+  ResolveWorkflowEscalationTool,
   ValidateWorkflowTool,
   PlanWorkflowGraphTool,
   GetExampleWorkflowTool,

packages/agents/src/tools/mcp-tools.ts

@@ -592,10 +592,31 @@ export class CreateWorkflowTool extends Tool {
   }
 }
 
+/**
+ * Escalated run payloads name the follow-up tool, so the model driving the
+ * loop knows how to answer without reading endpoint docs.
+ */
+function annotateEscalatedRun(run: unknown): unknown {
+  if (!run || typeof run !== "object") return run;
+  const record = run as Record<string, unknown>;
+  if (record["status"] !== "escalated") return run;
+  return {
+    ...record,
+    next_tool:
+      "A node invocation failed and the run is parked awaiting your verdict. " +
+      "Call resolve_workflow_escalation with this session_id and " +
+      "escalation_id and one of the escalation's allowedActions."
+  };
+}
+
 export class RunWorkflowTool extends Tool {
   readonly name = "run_workflow";
   readonly description =
-    "Execute a workflow with given parameters and return results.";
+    "Execute a workflow with given parameters and return results. With " +
+    "interactive=true a failing node invocation pauses the run and returns " +
+    "an escalation (status \"escalated\") for you to answer via " +
+    "resolve_workflow_escalation — retry, substitute, skip, or fail — " +
+    "instead of the whole run failing outright.";
   readonly jsonSchema = {
     type: "object" as const,
     properties: {
@@ -606,6 +627,12 @@ export class RunWorkflowTool extends Tool {
       params: {
         type: "object" as const,
         description: "Dictionary of input parameters for the workflow"
+      },
+      interactive: {
+        type: "boolean" as const,
+        description:
+          "Bubble node failures up as escalations you answer, instead of " +
+          "failing the run (default false)"
       }
     },
     required: ["workflow_id"]
@@ -615,9 +642,15 @@ export class RunWorkflowTool extends Tool {
     context: ProcessingContext,
     params: Record<string, unknown>
   ): Promise<unknown> {
-    return apiPost(context, `/api/workflows/${params["workflow_id"]}/run`, {
-      params: params["params"] ?? {}
-    });
+    const run = await apiPost(
+      context,
+      `/api/workflows/${params["workflow_id"]}/run`,
+      {
+        params: params["params"] ?? {},
+        ...(params["interactive"] === true ? { interactive: true } : {})
+      }
+    );
+    return annotateEscalatedRun(run);
   }
 
   userMessage(params: Record<string, unknown>): string {
@@ -653,7 +686,10 @@ export class DebugWorkflowTool extends Tool {
     "Run a workflow end-to-end and return a consolidated debug report: a " +
     "pass/fail verdict with the issues behind it, per-node status and errors, " +
     "logs, LLM calls, outputs, job record, and the workflow graph overview. " +
-    "Use this to troubleshoot a failing or misbehaving workflow and iterate.";
+    "Use this to troubleshoot a failing or misbehaving workflow and iterate. " +
+    "With interactive=true a failing node invocation pauses the run and " +
+    "returns an escalation (status \"escalated\") for you to answer via " +
+    "resolve_workflow_escalation before the report is produced.";
   readonly jsonSchema = {
     type: "object" as const,
     properties: {
@@ -665,6 +701,12 @@ export class DebugWorkflowTool extends Tool {
         type: "object" as const,
         description: "Input parameters keyed by input-node name"
       },
+      interactive: {
+        type: "boolean" as const,
+        description:
+          "Bubble node failures up as escalations you answer mid-run, " +
+          "instead of only reading them post-mortem (default false)"
+      },
       include_graph: {
         type: "boolean" as const,
         description:
@@ -697,7 +739,10 @@ export class DebugWorkflowTool extends Tool {
     const debugRun = await apiPostWithStatus(
       context,
       `/api/workflows/${workflowId}/debug`,
-      { params: params["params"] ?? {} }
+      {
+        params: params["params"] ?? {},
+        ...(params["interactive"] === true ? { interactive: true } : {})
+      }
     );
     const endpointMissing =
       !debugRun.ok && (debugRun.status === 404 || debugRun.status === 405);
@@ -708,6 +753,17 @@ export class DebugWorkflowTool extends Tool {
       });
     }
 
+    // An escalated run has produced no report yet — the job is parked on the
+    // failing node. Hand the escalation back for a verdict; the final report
+    // arrives from resolve_workflow_escalation once the run settles.
+    if (
+      run &&
+      typeof run === "object" &&
+      (run as Record<string, unknown>)["status"] === "escalated"
+    ) {
+      return { workflow_id: workflowId, run: annotateEscalatedRun(run) };
+    }
+
     const report: Record<string, unknown> = { workflow_id: workflowId, run };
     if (endpointMissing) {
       report["note"] =
@@ -732,6 +788,85 @@ export class DebugWorkflowTool extends Tool {
   }
 }
 
+export class ResolveWorkflowEscalationTool extends Tool {
+  readonly name = "resolve_workflow_escalation";
+  readonly description =
+    "Answer an escalation raised by an interactive run_workflow/debug_workflow " +
+    "run. The run is parked on the failing node until you decide: retry the " +
+    "invocation, substitute repaired outputs (only when the escalation carries " +
+    "a candidateOutput), skip the invocation, end the stream (streaming nodes), " +
+    "or fail the node. Only the escalation's allowedActions are accepted; the " +
+    "kernel enforces the same set. Returns the next escalation (answer it the " +
+    "same way) or the run's final report.";
+  readonly jsonSchema = {
+    type: "object" as const,
+    properties: {
+      session_id: {
+        type: "string" as const,
+        description: "The debug session id from the escalated response"
+      },
+      escalation_id: {
+        type: "string" as const,
+        description: "The escalation id being answered"
+      },
+      action: {
+        type: "string" as const,
+        enum: ["retry", "substitute", "skip", "end_stream", "fail"],
+        description: "The verdict — must be one of the escalation's allowedActions"
+      },
+      outputs: {
+        type: "object" as const,
+        description:
+          "For substitute: repaired output values keyed by the node's " +
+          "declared output slots"
+      },
+      reason: {
+        type: "string" as const,
+        description:
+          "For fail: a one-sentence reason surfaced as the run's error summary"
+      },
+      apply_to: {
+        type: "string" as const,
+        enum: ["invocation", "signature"],
+        description:
+          "For skip/fail: \"signature\" also resolves later failures with the " +
+          "same failureSignature without asking again (default \"invocation\")"
+      }
+    },
+    required: ["session_id", "escalation_id", "action"]
+  };
+
+  async process(
+    context: ProcessingContext,
+    params: Record<string, unknown>
+  ): Promise<unknown> {
+    const action = String(params["action"]);
+    const verdict: Record<string, unknown> = { action };
+    if (action === "substitute" && params["outputs"] !== undefined) {
+      verdict["outputs"] = params["outputs"];
+    }
+    if (action === "fail" && typeof params["reason"] === "string") {
+      verdict["reason"] = params["reason"];
+    }
+    if (
+      (action === "skip" || action === "fail") &&
+      typeof params["apply_to"] === "string"
+    ) {
+      verdict["applyTo"] = params["apply_to"];
+    }
+    const result = await apiPost(
+      context,
+      `/api/debug/sessions/${params["session_id"]}/verdict`,
+      { escalation_id: params["escalation_id"], verdict }
+    );
+    return annotateEscalatedRun(result);
+  }
+
+  userMessage(params: Record<string, unknown>): string {
+    return `Resolving workflow escalation with "${params["action"]}"`;
+  }
+}
+
 export class ValidateWorkflowTool extends Tool {
   readonly name = "validate_workflow";
   readonly description =
@@ -1441,6 +1576,7 @@ export function getAllMcpTools(options: GetAllMcpToolsOptions = {}): Tool[] {
     new CreateWorkflowTool(),
     new RunWorkflowTool(),
     new DebugWorkflowTool(),
+    new ResolveWorkflowEscalationTool(),
     new ValidateWorkflowTool(options.registry),
     new GetExampleWorkflowTool(),
     new ExportWorkflowDigraphTool(),

packages/agents/tests/mcp-tools.test.ts

@@ -8,6 +8,7 @@ import {
   CreateWorkflowTool,
   RunWorkflowTool,
   DebugWorkflowTool,
+  ResolveWorkflowEscalationTool,
   ValidateWorkflowTool,
   GetExampleWorkflowTool,
   ExportWorkflowDigraphTool,
@@ -335,6 +336,29 @@ describe("RunWorkflowTool", () => {
     expect(lastFetchUrl()).toContain("/api/workflows/wf-456/run");
     expect(lastFetchOpts().method).toBe("POST");
   });
+
+  it("forwards interactive and annotates an escalated response", async () => {
+    fetchSpy.mockResolvedValue({
+      ok: true,
+      json: async () => ({
+        status: "escalated",
+        session_id: "sess-1",
+        escalation_id: "esc-1",
+        escalation: { nodeId: "n1", allowedActions: ["skip", "fail"] }
+      }),
+      text: async () => ""
+    });
+
+    const result = (await tool.process(ctx, {
+      workflow_id: "wf-456",
+      interactive: true
+    })) as Record<string, unknown>;
+
+    const body = JSON.parse(lastFetchOpts().body as string);
+    expect(body.interactive).toBe(true);
+    expect(result.status).toBe("escalated");
+    expect(String(result.next_tool)).toContain("resolve_workflow_escalation");
+  });
 });
 
 describe("DebugWorkflowTool", () => {
@@ -393,6 +417,98 @@ describe("DebugWorkflowTool", () => {
     const urls = fetchSpy.mock.calls.map((c) => String(c[0]));
     expect(urls.some((u) => u.endsWith("/run"))).toBe(true);
   });
+
+  it("returns the escalation directly when the interactive run parks", async () => {
+    respondTo({
+      "/debug": {
+        body: {
+          status: "escalated",
+          session_id: "sess-9",
+          escalation_id: "esc-1",
+          escalation: { nodeId: "n1", allowedActions: ["skip", "fail"] }
+        }
+      }
+    });
+
+    const report = (await tool.process(ctx, {
+      workflow_id: "wf-1",
+      interactive: true
+    })) as Record<string, unknown>;
+
+    const body = JSON.parse(lastFetchOpts().body as string);
+    expect(body.interactive).toBe(true);
+    const run = report.run as Record<string, unknown>;
+    expect(run.status).toBe("escalated");
+    expect(String(run.next_tool)).toContain("resolve_workflow_escalation");
+    // No report exists yet, so neither the job nor the graph is fetched.
+    expect(fetchSpy.mock.calls).toHaveLength(1);
+  });
+});
+
+describe("ResolveWorkflowEscalationTool", () => {
+  const tool = new ResolveWorkflowEscalationTool();
+
+  it("posts the verdict to the session endpoint", async () => {
+    await tool.process(ctx, {
+      session_id: "sess-9",
+      escalation_id: "esc-1",
+      action: "skip",
+      apply_to: "signature"
+    });
+
+    expect(lastFetchUrl()).toContain("/api/debug/sessions/sess-9/verdict");
+    const body = JSON.parse(lastFetchOpts().body as string);
+    expect(body.escalation_id).toBe("esc-1");
+    expect(body.verdict).toEqual({ action: "skip", applyTo: "signature" });
+  });
+
+  it("carries substitute outputs and fail reasons, nothing else", async () => {
+    await tool.process(ctx, {
+      session_id: "s",
+      escalation_id: "e",
+      action: "substitute",
+      outputs: { value: "repaired" },
+      reason: "ignored for substitute"
+    });
+    const body = JSON.parse(lastFetchOpts().body as string);
+    expect(body.verdict).toEqual({
+      action: "substitute",
+      outputs: { value: "repaired" }
+    });
+
+    fetchSpy.mockClear();
+    await tool.process(ctx, {
+      session_id: "s",
+      escalation_id: "e",
+      action: "fail",
+      reason: "upstream data is unusable"
+    });
+    const failBody = JSON.parse(lastFetchOpts().body as string);
+    expect(failBody.verdict).toEqual({
+      action: "fail",
+      reason: "upstream data is unusable"
+    });
+  });
+
+  it("annotates a follow-up escalation so the loop continues", async () => {
+    fetchSpy.mockResolvedValue({
+      ok: true,
+      json: async () => ({
+        status: "escalated",
+        session_id: "sess-9",
+        escalation_id: "esc-2",
+        escalation: { nodeId: "n2", allowedActions: ["skip", "fail"] }
+      }),
+      text: async () => ""
+    });
+
+    const result = (await tool.process(ctx, {
+      session_id: "sess-9",
+      escalation_id: "esc-1",
+      action: "skip"
+    })) as Record<string, unknown>;
+    expect(String(result.next_tool)).toContain("resolve_workflow_escalation");
+  });
 });
 
 describe("ValidateWorkflowTool", () => {