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

Six defects found reviewing the execution/comparison stack:

1. Cancelled queued jobs could still execute. `ExecutionSession.create()`
   starts the kernel, and the WS runner checked the persisted "cancelled"
   status only afterwards, so the workflow ran (side effects included) and
   the branch never cancelled the session. Persistence now runs before the
   session is created. The regression test asserts the executor never ran,
   not just the DB row.

2. Golden fixtures never reached the verdict. `compareJourney` evaluated
   invariants and the cross-surface diff only, so every surface returning
   the same wrong answer passed. `core/golden.ts` compares terminal outputs
   and the normalized stream (per node-channel) against `expected/`, and
   fails the surface on a mismatch; faulted runs skip with a stated reason.
   `nodetool reliability update-goldens <journey>` recaptures fixtures from
   an unfaulted kernel run; linear-text-pipeline's stale shape is refreshed
   (constant-node frames now filtered by normalization, node_name is the
   node's name).

3. Nested payload differences vanished from the diff. `canonicalKey` used
   `JSON.stringify(msg, keys)`, whose array replacer is a key whitelist at
   every depth, so `{result:{output:"A"}}` and `{result:{output:"B"}}`
   serialized identically. Replaced with a recursive stable serializer.

4. Every production run built an unbounded, unread message queue.
   `ExecutionSession` always constructed `MessageStream`; no production
   caller iterates it. Capture is now opt-in (`captureMessages`) and
   bounded (`limits.messageBufferLimit`), and overflow throws instead of
   retaining.

5. `cleanup-leaks` measured nothing. No driver populated
   `resourceCounters`, and `checkLeaks` returned success on absence. The
   kernel driver now reads `ExecutionSession.resourceCounters()` (live
   actors, pending control responses, timers, bridge requests) and the
   ws-server driver reads the runner's slot counters after disconnect;
   a missing post-run snapshot is a violation.

6. The ws-server driver hid duplicate terminal frames. It dropped every
   repeated `job_update`, leaving "exactly one terminal update"
   undetectable on the one surface with a wire. Repeats are recorded and,
   only when they match this surface's documented shapes (eager
   running/cancel acks, authoritative terminal snapshot), tagged
   `redundant` — discounted by normalization and terminal-uniqueness.
   Any other duplicate now fails.

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

Author: claude

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[] = [];