fix(reliability,execution): make cancellation, goldens, diffs, and leak accounting real (#4630)

Author: georgi

docs/RELIABILITY_ARCHITECTURE.md

@@ -169,6 +169,13 @@ Rules that keep journeys trustworthy:
   not globally, because the actor model legitimately interleaves.
 - **Every journey states its invariant set explicitly** (§6) and the
   harness refuses a journey that asserts nothing beyond "completed".
+- **A declared golden decides the verdict.** `assertions.outputs` /
+  `assertions.streamShape` compare every surface's record against
+  `expected/` and fail the journey on a mismatch — invariants and the
+  cross-surface diff both pass when every surface returns the same wrong
+  answer. Faulted runs skip the comparison (and say so): a golden
+  describes the unfaulted contract. `nodetool reliability update-goldens
+  <journey>` recaptures the fixtures from an unfaulted kernel run.
 
 ## 5. Which workflows belong in the suite
 
@@ -234,7 +241,11 @@ the checks that convert today's advisory logging into failures:
 - Terminal precedence is exactly cancel > suspend > failed > completed
   (`runner.ts:625`) — asserted, not assumed.
 - Exactly one terminal `job_update` per run, and nothing follows it on
-  that job's stream.
+  that job's stream. A driver records repeated frames rather than
+  dropping them; a repeat its surface documents (the ws-server's eager
+  `running`/`cancelled` acks and its authoritative terminal snapshot) is
+  tagged `redundant` and discounted here and in the cross-surface diff.
+  Any other duplicate is untagged, and fails.
 
 **Cleanup and leaks.**
 - `_checkPendingInboxWork` finding pending work is a journey **failure**,
@@ -247,6 +258,11 @@ the checks that convert today's advisory logging into failures:
   repeated journey iterations (leak journeys run N=20 and compare).
 - Cancellation completes within a budget (e.g. 5 s) and `finalize()` ran
   for every started actor.
+- The counters above are **measured** — the kernel driver reads them off
+  `ExecutionSession.resourceCounters()`, the ws-server driver off the
+  runner's `slotCounters` once the connection is gone. A journey that
+  declares `cleanup-leaks` on a surface whose driver produces no post-run
+  snapshot fails on that: "nothing measured" must not read as "no leaks".
 
 **Determinism.**
 - Two runs of the same journey with the same cassettes produce identical

packages/cli/src/commands/reliability.ts

@@ -70,6 +70,25 @@ export function registerReliabilityCommands(program: Command): void {
       }
     });
 
+  reliability
+    .command("update-goldens <journey>")
+    .description(
+      "Rewrite a journey's expected/ fixtures from a fresh unfaulted kernel run — " +
+        "for a golden that legitimately moved. Review the diff before committing"
+    )
+    .action(async (journeyName: string) => {
+      try {
+        const { updateJourneyGoldens } = await import(
+          "@nodetool-ai/reliability-harness"
+        );
+        const { written } = await updateJourneyGoldens(journeyName);
+        for (const file of written) console.log(`wrote ${file}`);
+      } catch (e) {
+        console.error(String(e));
+        process.exit(1);
+      }
+    });
+
   reliability
     .command("list")
     .description("List the journeys under reliability/journeys/")

packages/execution/README.md

@@ -81,8 +81,14 @@ const session = await ExecutionSession.create({
   persistence: { onAccepted, onTerminal } | null,
   limits: { runTimeoutMs, bufferLimit },  // nodeTimeoutMs: not yet supported, see above
   requireTerminalResult: true,  // optional — output-name rewriting
+  captureMessages: true,        // optional — required to read `messages` below
 });
 
+// Only with `captureMessages: true`, and only if you actually drain it:
+// capture queues each message until a consumer pulls it, so a host that just
+// awaits `result` would hold the whole run's messages (audio chunks included)
+// for nothing. Reading `messages` without the flag throws; a consumer that
+// falls behind `limits.messageBufferLimit` makes the iterator throw.
 for await (const message of session.messages) { /* ProcessingMessage */ }
 
 await session.pushInput("input_name", value);

packages/execution/src/index.ts

@@ -3,6 +3,7 @@
  */
 export { ExecutionSession } from "./session.js";
 export { normalizeGraph } from "./normalize-graph.js";
+export { DEFAULT_MESSAGE_BUFFER_LIMIT } from "./message-stream.js";
 export type {
   BridgeFactory,
   Edge,

packages/execution/src/message-stream.ts

@@ -3,18 +3,40 @@
  * `AsyncIterable<ProcessingMessage>`. Modeled on the queue/wake pattern in
  * `packages/workflow-runner/src/run.ts`'s `runWorkflow` generator — the one
  * other place in the repo already streams a kernel run as an async sequence.
+ *
+ * The queue is bounded. A caller that opts into capture (see
+ * `ExecutionSessionOptions.captureMessages`) but never iterates would
+ * otherwise retain every message of the run — including real-time audio
+ * chunks, which the kernel's own message retention deliberately avoids
+ * holding (roughly 25 MB/minute). Past the limit the stream stops queueing
+ * and the iterator throws, so a non-draining consumer fails loudly instead of
+ * growing until the run ends.
  */
 import type { ProcessingContext } from "@nodetool-ai/runtime";
 import type { ProcessingMessage } from "@nodetool-ai/protocol";
 
+/** Queued messages allowed to pile up before the stream gives up. */
+export const DEFAULT_MESSAGE_BUFFER_LIMIT = 10_000;
+
 export class MessageStream implements AsyncIterable<ProcessingMessage> {
   private readonly queue: ProcessingMessage[] = [];
   private waiter: (() => void) | null = null;
   private closed = false;
+  private overflowedAt: number | null = null;
+  private readonly limit: number;
   private readonly unsubscribe: () => void;
 
-  constructor(context: ProcessingContext) {
+  /** `limit` of 0 (or a negative number) means unbounded. */
+  constructor(context: ProcessingContext, limit = DEFAULT_MESSAGE_BUFFER_LIMIT) {
+    this.limit = limit > 0 ? limit : Number.POSITIVE_INFINITY;
     this.unsubscribe = context.addMessageListener((message) => {
+      if (this.queue.length >= this.limit) {
+        if (this.overflowedAt === null) {
+          this.overflowedAt = this.queue.length;
+          this.unsubscribe();
+        }
+        return;
+      }
       this.queue.push(message);
       this.wake();
     });
@@ -39,6 +61,15 @@ export class MessageStream implements AsyncIterable<ProcessingMessage> {
       while (this.queue.length > 0) {
         yield this.queue.shift()!;
       }
+      if (this.overflowedAt !== null) {
+        throw new Error(
+          `ExecutionSession message stream overflowed after ${this.overflowedAt} ` +
+            `queued messages (limit ${this.limit}) — the consumer fell behind, ` +
+            "so messages past that point were dropped rather than retained. " +
+            "Iterate `session.messages` as the run progresses, or raise " +
+            "`limits.messageBufferLimit`."
+        );
+      }
       if (this.closed) return;
       await new Promise<void>((resolve) => {
         this.waiter = resolve;

packages/execution/src/session.ts

@@ -30,8 +30,10 @@ export class ExecutionSession {
   readonly graph: HydratedGraphData;
 
   private readonly runner: WorkflowRunner;
-  private readonly stream: MessageStream;
+  /** Null unless the caller opted into `captureMessages` (see options). */
+  private readonly stream: MessageStream | null;
   private readonly persistence: ExecutionSessionOptions["persistence"];
+  private readonly bridge: { pendingRequestCount?: number } | null;
   private readonly resultPromise: Promise<RunResult>;
   private runTimeoutHandle: ReturnType<typeof setTimeout> | null = null;
   private _cancelReason: string | null = null;
@@ -45,15 +47,21 @@ export class ExecutionSession {
     persistence: ExecutionSessionOptions["persistence"];
     params: Record<string, unknown>;
     triggerEvent: ExecutionSessionOptions["triggerEvent"];
+    bridge: { pendingRequestCount?: number } | null;
     closeBridge: () => void;
     runTimeoutMs: number | undefined;
+    captureMessages: boolean;
+    messageBufferLimit: number | undefined;
   }) {
     this.jobId = init.jobId;
     this.workflowId = init.workflowId;
     this.graph = init.graph;
     this.runner = init.runner;
     this.persistence = init.persistence;
-    this.stream = new MessageStream(init.context);
+    this.bridge = init.bridge;
+    this.stream = init.captureMessages
+      ? new MessageStream(init.context, init.messageBufferLimit)
+      : null;
 
     if (init.runTimeoutMs && init.runTimeoutMs > 0) {
       this.runTimeoutHandle = setTimeout(() => {
@@ -80,7 +88,7 @@ export class ExecutionSession {
         // The terminal message was emitted synchronously before run()
         // resolved (ProcessingContext.emit() calls listeners inline), so the
         // stream can close now without dropping it.
-        this.stream.close();
+        this.stream?.close();
       });
 
     this.resultPromise
@@ -206,13 +214,28 @@ export class ExecutionSession {
       persistence: options.persistence ?? null,
       params: options.params ?? {},
       triggerEvent: options.triggerEvent ?? null,
+      bridge,
       closeBridge,
-      runTimeoutMs: options.limits?.runTimeoutMs
+      runTimeoutMs: options.limits?.runTimeoutMs,
+      captureMessages: options.captureMessages === true,
+      messageBufferLimit: options.limits?.messageBufferLimit
     });
   }
 
-  /** Validated, live message stream — closes once the run reaches a terminal state. */
+  /**
+   * Live message stream — closes once the run reaches a terminal state.
+   * Requires `captureMessages: true` at `create()`; without it nothing is
+   * queued (see that option) and reading this throws rather than handing back
+   * a stream that would silently yield nothing.
+   */
   get messages(): AsyncIterable<ProcessingMessage> {
+    if (!this.stream) {
+      throw new Error(
+        "ExecutionSession: `messages` requires `captureMessages: true` at " +
+          "create() — message capture is opt-in so a host that only awaits " +
+          "`result` never queues a run's messages unread."
+      );
+    }
     return this.stream;
   }
 
@@ -221,6 +244,31 @@ export class ExecutionSession {
     return this.resultPromise;
   }
 
+  /**
+   * Live resource counts, for leak accounting: after a terminal result every
+   * one of these must be back to zero. Measured, not inferred — the
+   * reliability harness's `cleanup-leaks` invariant asserts against these
+   * numbers and reports a violation when a driver can't produce them.
+   */
+  resourceCounters(): {
+    liveActors: number;
+    pendingControlResponses: number;
+    pendingTimers: number;
+    pythonBridgePendingRequests: number;
+  } {
+    return {
+      liveActors: this.runner.liveActorCount,
+      pendingControlResponses: this.runner.pendingControlResponseCount,
+      // The session's own run-timeout timer is the only timer it owns; it is
+      // cleared when the run settles.
+      pendingTimers: this.runTimeoutHandle === null ? 0 : 1,
+      pythonBridgePendingRequests:
+        typeof this.bridge?.pendingRequestCount === "number"
+          ? this.bridge.pendingRequestCount
+          : 0
+    };
+  }
+
   /** The reason passed to the most recent `cancel()` call, if any. */
   get cancelReason(): string | null {
     return this._cancelReason;

packages/execution/src/types.ts

@@ -62,6 +62,14 @@ export interface ExecutionLimits {
   nodeTimeoutMs?: number;
   /** Forwarded to `WorkflowRunnerOptions.bufferLimit` (per-inbox cap). */
   bufferLimit?: number | null;
+  /**
+   * Cap on messages queued for an un-drained `session.messages` consumer
+   * (default {@link DEFAULT_MESSAGE_BUFFER_LIMIT}); `0` disables the cap. Only
+   * meaningful together with `captureMessages`. Past the cap the stream stops
+   * queueing and the iterator throws — a consumer that fell behind fails
+   * loudly instead of retaining the whole run's message history.
+   */
+  messageBufferLimit?: number;
 }
 
 export interface ExecutionSessionOptions {
@@ -144,6 +152,17 @@ export interface ExecutionSessionOptions {
    * surface and the one place strict mode is meant to be on by default.
    */
   strict?: boolean;
+  /**
+   * Capture every emitted message into `session.messages` (default `false`).
+   * Off by default because capture is retention: the queue holds each message
+   * until a consumer pulls it, and a host that only awaits `session.result`
+   * (every production caller today) would grow one unread queue per run — the
+   * kernel is explicit about not retaining real-time chunks for exactly that
+   * reason. Turn it on only when actually iterating the stream (the
+   * reliability harness's kernel driver does); see `limits.messageBufferLimit`
+   * for the cap that catches a consumer falling behind.
+   */
+  captureMessages?: boolean;
 }
 
 export type { NodeDescriptor, Edge, ProcessingMessage, RunResult };

packages/execution/tests/message-capture.test.ts

@@ -0,0 +1,61 @@
+/**
+ * Message capture is opt-in and bounded: a host that only awaits
+ * `session.result` (every production caller) must not accumulate an unread
+ * queue of the run's messages, and a consumer that opts in but falls behind
+ * must fail loudly instead of retaining everything.
+ */
+import { describe, it, expect } from "vitest";
+import { ExecutionSession } from "../src/index.js";
+import { buildTestRegistry } from "./fixtures.js";
+
+const NO_BRIDGE = async () => null;
+
+const DOUBLE_GRAPH = {
+  nodes: [
+    { id: "v", type: "nodetool.input.Value", properties: {} },
+    { id: "double", type: "test.execution.Double", properties: {} }
+  ],
+  edges: [
+    { source: "v", sourceHandle: "output", target: "double", targetHandle: "value" }
+  ]
+};
+
+describe("ExecutionSession — message capture", () => {
+  it("queues nothing and refuses `messages` when capture is not requested", async () => {
+    const registry = buildTestRegistry();
+    const session = await ExecutionSession.create({
+      graph: DOUBLE_GRAPH,
+      registry,
+      bridgeFactory: NO_BRIDGE,
+      params: { v: 5 }
+    });
+
+    expect(() => session.messages).toThrow(/captureMessages/);
+    const result = await session.result;
+    expect(result.status).toBe("completed");
+  });
+
+  it("throws once an un-drained consumer exceeds the buffer limit", async () => {
+    const registry = buildTestRegistry();
+    const session = await ExecutionSession.create({
+      graph: {
+        nodes: [{ id: "loop", type: "test.execution.Loop", properties: {} }],
+        edges: []
+      },
+      registry,
+      bridgeFactory: NO_BRIDGE,
+      captureMessages: true,
+      limits: { runTimeoutMs: 500, messageBufferLimit: 3 }
+    });
+
+    // Let the run emit well past the limit before pulling anything.
+    await session.result;
+
+    const drain = async (): Promise<void> => {
+      for await (const _message of session.messages) {
+        // drain
+      }
+    };
+    await expect(drain()).rejects.toThrow(/overflowed/);
+  });
+});

packages/execution/tests/session-run.test.ts

@@ -105,7 +105,8 @@ describe("ExecutionSession — messages", () => {
       },
       registry,
       bridgeFactory: NO_BRIDGE,
-      params: { v: 5 }
+      params: { v: 5 },
+      captureMessages: true
     });
 
     const seen: string[] = [];