Address review: consistent failed-run payload, sanitized bounds, no waiter leak

- A thrown interactive run now reports through buildWorkflowRunPayload, so
  the debug surface keeps its summary + verdict shape on hard failures.
- max_decisions / max_retries_per_node / decision_timeout_ms are API
  inputs: anything but a sane integer falls back to the default instead of
  reaching BoundedHandle (a NaN timeout failed every decision instantly).
- waitForEvent's escalation subscription is cancellable, so the done branch
  winning the race no longer strands a waiter per verdict round trip.

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

Author: claude

packages/websocket/src/debug-sessions.ts

@@ -39,6 +39,12 @@ interface PendingEscalation {
   settle: (outcome: DecisionOutcome) => void;
 }
 
+/** What escalation subscribers see: the record, never the settle handle. */
+export interface ParkedEscalation {
+  id: string;
+  escalation: Escalation;
+}
+
 /**
  * A `SupervisorHandle` that answers nothing itself: each `decide()` parks the
  * escalation until `submit()` delivers a verdict (or the decision signal
@@ -47,7 +53,7 @@ interface PendingEscalation {
  */
 export class InteractiveEscalationHandle implements SupervisorHandle {
   private _pending: PendingEscalation | null = null;
-  private _waiters: Array<(entry: PendingEscalation) => void> = [];
+  private _waiters: Array<(entry: ParkedEscalation) => void> = [];
   private _counter = 0;
   private _closed = false;
 
@@ -81,19 +87,18 @@ export class InteractiveEscalationHandle implements SupervisorHandle {
     return { id: this._pending.id, escalation: this._pending.escalation };
   }
 
-  /** Resolves with the next parked escalation (immediately if one is parked). */
-  nextEscalation(): Promise<{ id: string; escalation: Escalation }> {
-    if (this._pending) {
-      return Promise.resolve({
-        id: this._pending.id,
-        escalation: this._pending.escalation
-      });
-    }
-    return new Promise((resolve) => {
-      this._waiters.push((entry) =>
-        resolve({ id: entry.id, escalation: entry.escalation })
-      );
-    });
+  /**
+   * Subscribe to the next parked escalation. Returns an unsubscribe so a
+   * caller racing this against run completion can withdraw when the run wins
+   * — otherwise every verdict round trip would strand one waiter until the
+   * session is swept.
+   */
+  subscribe(waiter: (entry: ParkedEscalation) => void): () => void {
+    this._waiters.push(waiter);
+    return () => {
+      const index = this._waiters.indexOf(waiter);
+      if (index !== -1) this._waiters.splice(index, 1);
+    };
   }
 
   /**
@@ -178,18 +183,33 @@ export class DebugSession {
    * parked decision fails closed on its own timeout, so this always resolves.
    */
   async waitForEvent(): Promise<DebugSessionEvent> {
-    return Promise.race([
-      this._done.then(
-        (report): DebugSessionEvent => ({ kind: "done", report })
-      ),
-      this._handle.nextEscalation().then(
-        (entry): DebugSessionEvent => ({
-          kind: "escalated",
-          escalationId: entry.id,
-          escalation: entry.escalation
+    const pending = this._handle.current();
+    if (pending) {
+      return {
+        kind: "escalated",
+        escalationId: pending.id,
+        escalation: pending.escalation
+      };
+    }
+    let unsubscribe: () => void = () => {};
+    try {
+      return await Promise.race([
+        this._done.then(
+          (report): DebugSessionEvent => ({ kind: "done", report })
+        ),
+        new Promise<DebugSessionEvent>((resolve) => {
+          unsubscribe = this._handle.subscribe((entry) =>
+            resolve({
+              kind: "escalated",
+              escalationId: entry.id,
+              escalation: entry.escalation
+            })
+          );
         })
-      )
-    ]);
+      ]);
+    } finally {
+      unsubscribe();
+    }
   }
 
   /** Current state without waiting. */

packages/websocket/src/http-api.ts

@@ -884,18 +884,18 @@ export async function handleWorkflowRun(
       // guarantees an LLM supervisor gets — decision/retry caps, a per-decision
       // timeout that fails closed, sticky verdicts — just with a timeout sized
       // for an agent's tool round trip instead of one model call.
+      // These are API inputs: anything but a sane integer (NaN, Infinity,
+      // negatives, fractions) falls back to the default rather than reaching
+      // `BoundedHandle` — a NaN timeout would fail every decision instantly.
+      const maxDecisions = boundedRunOption(body?.max_decisions, 1);
+      const maxRetriesPerNode = boundedRunOption(body?.max_retries_per_node, 0);
       interactiveHandle = new InteractiveEscalationHandle();
       supervisorHandle = new BoundedHandle(interactiveHandle, {
-        ...(typeof body?.max_decisions === "number"
-          ? { maxDecisions: body.max_decisions }
-          : {}),
-        ...(typeof body?.max_retries_per_node === "number"
-          ? { maxRetriesPerNode: body.max_retries_per_node }
-          : {}),
+        ...(maxDecisions !== undefined ? { maxDecisions } : {}),
+        ...(maxRetriesPerNode !== undefined ? { maxRetriesPerNode } : {}),
         decisionTimeoutMs:
-          typeof body?.decision_timeout_ms === "number"
-            ? body.decision_timeout_ms
-            : INTERACTIVE_DECISION_TIMEOUT_MS
+          boundedRunOption(body?.decision_timeout_ms, 1) ??
+          INTERACTIVE_DECISION_TIMEOUT_MS
       });
     }
     runner = new WorkflowRunner(job.id, {
@@ -983,12 +983,17 @@ export async function handleWorkflowRun(
           error: String(saveError)
         });
       }
-      return {
-        job_id: job.id,
-        workflow_id: workflowId,
+      // Keep the payload shape of a failed run — the debug surface still gets
+      // a summary and verdict — instead of a bare {status, error} object.
+      const failed: WorkflowRunResult = {
         status: "failed",
-        error: message
+        error: message,
+        messages: [],
+        outputs: {}
       };
+      return buildWorkflowRunPayload(job.id, workflowId, failed, debug, {
+        background: false
+      });
     } finally {
       supervisorHandle?.close();
     }
@@ -1008,6 +1013,13 @@ export async function handleWorkflowRun(
 
 type WorkflowRunResult = Awaited<ReturnType<WorkflowRunner["run"]>>;
 
+/** An interactive-run bound from the request body: an integer ≥ min, or undefined. */
+function boundedRunOption(value: unknown, min: number): number | undefined {
+  return typeof value === "number" && Number.isInteger(value) && value >= min
+    ? value
+    : undefined;
+}
+
 async function finalizeWorkflowRunJob(
   job: Job,
   result: WorkflowRunResult