From 1ab816daad35e5fd37a05ed90d1955b421d29cd8 Mon Sep 17 00:00:00 2001
From: Daniel Bannert <d.bannert@anolilab.de>
Date: Tue, 20 Oct 2026 12:46:09 +0200
Subject: [PATCH 1/2] fix(vis): keep the CI log alive when the runner stops
 draining stdout
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A CI run's task-output dump was one unbounded process.stdout.write of
the whole buffered task output (up to 256 KiB). When a runner stops
draining the stdout pipe mid-dump, that single write wedges, every later
line queues invisibly behind it, and the log freezes mid-line — observed
in the wild as a 47-minute silent hang of an otherwise green test leg
that had to be cancelled by hand.

Route all CI-path output (status lines, group dumps, summaries) through
a bounded, backpressure-aware writer:

- writes are chunked (<= 32 KiB, line-aligned, surrogate-pair safe) and
  honor backpressure: after a false return the pump waits for drain
  before committing the next chunk, so at most one chunk ever sits
  inside the stream's buffer
- the fast path stays synchronous: a healthy reader sees identical
  behavior to the direct writes this replaces
- a watchdog escalates a stuck channel: warn on stderr (naming the
  pending task), then destroy the wedged stdout and reroute the
  remaining output to stderr (at most one chunk lost), then — only if
  stderr is wedged too — drop output so the run can still finish
- stream errors (EPIPE and friends) reroute immediately instead of
  crashing or wedging the log
---
 .../__tests__/tui/ci-output-writer.test.ts    | 367 ++++++++++++
 .../tooling/vis/src/tui/ci-output-writer.ts   | 528 ++++++++++++++++++
 .../tooling/vis/src/tui/static-life-cycle.ts  |  24 +-
 packages/tooling/vis/src/tui/status-utils.ts  |  37 +-
 .../tooling/vis/src/tui/summary-life-cycle.ts |   3 +-
 5 files changed, 934 insertions(+), 25 deletions(-)
 create mode 100644 packages/tooling/vis/__tests__/tui/ci-output-writer.test.ts
 create mode 100644 packages/tooling/vis/src/tui/ci-output-writer.ts

diff --git a/packages/tooling/vis/__tests__/tui/ci-output-writer.test.ts b/packages/tooling/vis/__tests__/tui/ci-output-writer.test.ts
new file mode 100644
index 00000000..ca0721c2
--- /dev/null
+++ b/packages/tooling/vis/__tests__/tui/ci-output-writer.test.ts
@@ -0,0 +1,367 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import type { CiOutputStream } from "../../src/tui/ci-output-writer";
+import { createCiOutputWriter, flushCiOutput, splitCiOutputChunks, writeCiOutput } from "../../src/tui/ci-output-writer";
+
+/**
+ * Poll `condition` until it holds. Polls instead of `vi.waitFor` so every
+ * `expect` in a test runs exactly once and `expect.assertions` counts stay
+ * exact.
+ */
+const waitForCondition = async (condition: () => boolean, timeoutMs = 2000): Promise<void> => {
+    const deadline = Date.now() + timeoutMs;
+
+    while (!condition()) {
+        if (Date.now() > deadline) {
+            throw new Error("waitForCondition: condition not met within the timeout");
+        }
+
+        await new Promise((resolve) => {
+            setTimeout(resolve, 5);
+        });
+    }
+};
+
+/** The events the fake stream's listener registry supports. */
+type FakeStreamEvent = "close" | "drain" | "error";
+type FakeStreamEventCloseOrError = "close" | "error";
+
+/**
+ * A controllable stand-in for `process.stdout` / `process.stderr`.
+ * `accepting` toggles backpressure: `write()` records every chunk either
+ * way and returns `false` while the pipe is "full", just like a real
+ * stream whose internal buffer is over the high-water mark.
+ */
+const createFakeStream = () => {
+    const closeListeners = new Set<(error?: Error) => void>();
+    const drainListeners = new Set<(error?: Error) => void>();
+    const errorListeners = new Set<(error?: Error) => void>();
+
+    const state = {
+        accepting: true,
+        destroyCount: 0,
+        destroyed: false,
+        writes: [] as string[],
+    };
+
+    const emit = (event: FakeStreamEvent, error?: Error): void => {
+        const listeners = event === "close" ? closeListeners : event === "drain" ? drainListeners : errorListeners;
+
+        for (const listener of listeners) {
+            listener(error);
+        }
+    };
+
+    const stream = {
+        destroy: (): void => {
+            state.destroyed = true;
+            state.destroyCount += 1;
+        },
+        off(event: FakeStreamEvent, listener: (error?: Error) => void): unknown {
+            if (event === "close") {
+                closeListeners.delete(listener);
+            } else if (event === "drain") {
+                drainListeners.delete(listener);
+            } else {
+                errorListeners.delete(listener);
+            }
+
+            return stream;
+        },
+        on(event: FakeStreamEventCloseOrError, listener: (error?: Error) => void): unknown {
+            if (event === "close") {
+                closeListeners.add(listener);
+            } else {
+                errorListeners.add(listener);
+            }
+
+            return stream;
+        },
+        once(event: FakeStreamEvent, listener: (error?: Error) => void): unknown {
+            if (event === "close") {
+                closeListeners.add(listener);
+            } else if (event === "drain") {
+                drainListeners.add(listener);
+            } else {
+                errorListeners.add(listener);
+            }
+
+            return stream;
+        },
+        write(chunk: string): boolean {
+            state.writes.push(chunk);
+
+            return state.accepting && !state.destroyed;
+        },
+    };
+
+    return {
+        close: (): void => {
+            emit("close");
+        },
+        get destroyCount(): number {
+            return state.destroyCount;
+        },
+        drain: (): void => {
+            emit("drain");
+        },
+        error: (code: string): void => {
+            const error = new Error(`write ${code}`) as Error & { code: string };
+
+            error.code = code;
+            emit("error", error);
+        },
+        setAccepting: (accepting: boolean): void => {
+            state.accepting = accepting;
+        },
+        stream: stream as CiOutputStream,
+        get writes(): string[] {
+            return state.writes;
+        },
+    };
+};
+
+/** Thresholds small enough for real timers, still ordered: warn < reroute < drop. */
+const fastThresholds = { chunkCodeUnits: 64, stallDropMs: 150, stallRerouteMs: 100, stallWarnMs: 50 };
+
+describe(splitCiOutputChunks, () => {
+    it("returns small text as a single chunk", () => {
+        expect.assertions(1);
+
+        expect(splitCiOutputChunks("hello", 64)).toStrictEqual(["hello"]);
+    });
+
+    it("caps every chunk at the limit and rejoins losslessly", () => {
+        expect.assertions(2);
+
+        const text = "a".repeat(200);
+
+        const chunks = splitCiOutputChunks(text, 64);
+
+        expect(chunks.every((chunk) => chunk.length <= 64)).toBe(true);
+        expect(chunks.join("")).toBe(text);
+    });
+
+    it("prefers a line boundary inside the window", () => {
+        expect.assertions(2);
+
+        const text = `${"a".repeat(40)}\n${"b".repeat(40)}`;
+
+        const chunks = splitCiOutputChunks(text, 50);
+
+        expect(chunks).toStrictEqual([`${"a".repeat(40)}\n`, "b".repeat(40)]);
+        expect(chunks.every((chunk) => chunk.length <= 50)).toBe(true);
+    });
+
+    it("never splits a surrogate pair at a chunk boundary", () => {
+        expect.assertions(2);
+
+        // A 4-byte emoji (two UTF-16 code units) straddling the 64-unit window.
+        const text = `${"a".repeat(63)}😀${"b".repeat(10)}`;
+
+        const chunks = splitCiOutputChunks(text, 64);
+
+        // No chunk may end on an unpaired high surrogate — that would encode
+        // the pair's halves as replacement characters and corrupt the output.
+        // A trailing lone surrogate reads as its own code point (< 0x10000).
+        const endsOnHighSurrogate = chunks.some((chunk) => {
+            const last = chunk.codePointAt(chunk.length - 1) ?? 0;
+
+            return last >= 0xd8_00 && last <= 0xdb_ff;
+        });
+
+        expect(endsOnHighSurrogate).toBe(false);
+        expect(chunks.join("")).toBe(text);
+    });
+});
+
+describe(createCiOutputWriter, () => {
+    it("writes through to a healthy stream synchronously, in order", () => {
+        expect.assertions(4);
+
+        const stdout = createFakeStream();
+        const stderr = createFakeStream();
+        const writer = createCiOutputWriter(stdout.stream, stderr.stream, fastThresholds);
+
+        writer.write("first\n", "a");
+        writer.write("second\n", "b");
+
+        // Fast path: no await needed — the writes already landed.
+        expect(stdout.writes.join("")).toBe("first\nsecond\n");
+        expect(writer.status()).toStrictEqual({ channel: "stdout", pendingChunks: 0, pendingCodeUnits: 0 });
+        expect(stderr.writes).toStrictEqual([]);
+        expect(stdout.destroyCount).toBe(0);
+    });
+
+    it("waits for drain before committing the next chunk and keeps order", async () => {
+        expect.assertions(3);
+
+        const stdout = createFakeStream();
+        const stderr = createFakeStream();
+        const writer = createCiOutputWriter(stdout.stream, stderr.stream, fastThresholds);
+
+        stdout.setAccepting(false);
+
+        writer.write(`${"a".repeat(62)}\n${"b".repeat(62)}\n`, "task");
+
+        // Chunk one is committed to the wedged stream; chunk two must wait.
+        expect(stdout.writes).toHaveLength(1);
+
+        const flushed = writer.flush();
+
+        stdout.setAccepting(true);
+        stdout.drain();
+
+        await flushed;
+
+        expect(stdout.writes.join("")).toBe(`${"a".repeat(62)}\n${"b".repeat(62)}\n`);
+        expect(writer.status().channel).toBe("stdout");
+    });
+
+    it("reroutes to stderr when stdout stalls, losing at most one chunk — the observed CI hang", async () => {
+        expect.assertions(6);
+
+        const stdout = createFakeStream();
+        const stderr = createFakeStream();
+        const writer = createCiOutputWriter(stdout.stream, stderr.stream, fastThresholds);
+
+        // The runner stopped draining our stdout: every write is backpressured.
+        stdout.setAccepting(false);
+
+        const dump = "line\n".repeat(40);
+
+        writer.write(dump, "runtime:test:coverage");
+        writer.write("✓ runtime:test:coverage (1 m 22 s)\n", "runtime:test:coverage");
+
+        // Exactly one bounded chunk was handed to the wedged stdout.
+        expect(stdout.writes).toHaveLength(1);
+        expect(stdout.writes[0]?.length).toBeLessThanOrEqual(fastThresholds.chunkCodeUnits);
+
+        // The reroute lands on the still-flowing stderr pipe.
+        await waitForCondition(() => writer.status().channel === "stderr");
+
+        const stderrText = stderr.writes.join("");
+
+        // The stall was reported, naming what is stuck…
+        expect(stderrText).toContain("stalled");
+        expect(stderrText).toContain("runtime:test:coverage");
+        // …and the remaining output survived, in order, ending with the status line.
+        expect(stderrText.indexOf("rerouting")).toBeGreaterThan(-1);
+        expect(stderrText.endsWith("✓ runtime:test:coverage (1 m 22 s)\n")).toBe(true);
+
+        await writer.flush();
+    });
+
+    it("drops output entirely when both pipes stall, so the run can still finish", async () => {
+        expect.assertions(4);
+
+        const stdout = createFakeStream();
+        const stderr = createFakeStream();
+        const writer = createCiOutputWriter(stdout.stream, stderr.stream, fastThresholds);
+
+        stdout.setAccepting(false);
+        stderr.setAccepting(false);
+
+        // Multi-chunk so the reroute leaves queued output poking stderr —
+        // the stderr stall clock (and the drop rung) only runs with pending work.
+        writer.write("line\n".repeat(40), "task");
+
+        await waitForCondition(() => writer.status().channel === "dropped");
+
+        // flush() must resolve on the dropped channel — no caller can hang on it.
+        await writer.flush();
+
+        writer.write("late output", "task");
+
+        expect(writer.status()).toStrictEqual({ channel: "dropped", pendingChunks: 0, pendingCodeUnits: 0 });
+        expect(stdout.destroyCount).toBe(1);
+        expect(stderr.destroyCount).toBe(1);
+        expect(stderr.writes.length).toBeGreaterThan(0);
+    });
+
+    it("reroutes immediately when stdout errors, without waiting for the stall window", async () => {
+        expect.assertions(4);
+
+        const stdout = createFakeStream();
+        const stderr = createFakeStream();
+        const writer = createCiOutputWriter(stdout.stream, stderr.stream, { ...fastThresholds, stallRerouteMs: 60_000, stallWarnMs: 60_000 });
+
+        writer.write("before the error\n", "task");
+        stdout.error("EPIPE");
+        writer.write("after the error\n", "task");
+
+        expect(writer.status().channel).toBe("stderr");
+        expect(stderr.writes.join("")).toContain("EPIPE");
+        expect(stderr.writes.join("")).toContain("after the error");
+        expect(stdout.writes.join("")).toContain("before the error");
+
+        await writer.flush();
+    });
+
+    it("does not escalate while the reader keeps making progress", async () => {
+        expect.assertions(3);
+
+        const stdout = createFakeStream();
+        const stderr = createFakeStream();
+        const writer = createCiOutputWriter(stdout.stream, stderr.stream, fastThresholds);
+
+        // A slow-but-alive reader: every write is backpressured, but drain
+        // arrives well inside the stall window. Recursive (not a loop) so
+        // no `await` sits inside a loop body.
+        const drainSoon = async (remaining: number): Promise<void> => {
+            if (remaining === 0) {
+                return;
+            }
+
+            await new Promise((resolve) => {
+                setTimeout(resolve, 10);
+            });
+
+            stdout.drain();
+
+            await drainSoon(remaining - 1);
+        };
+
+        // Backpressure from the first chunk on.
+        stdout.setAccepting(false);
+
+        writer.write("chunk\n".repeat(20), "task");
+
+        const flushed = writer.flush();
+
+        await drainSoon(4);
+
+        stdout.setAccepting(true);
+        stdout.drain();
+
+        await flushed;
+
+        // Slow, but everything landed on stdout and no escalation fired.
+        expect(writer.status().channel).toBe("stdout");
+        expect(stderr.writes).toStrictEqual([]);
+        expect(stdout.writes.join("")).toContain("chunk");
+    });
+});
+
+describe(writeCiOutput, () => {
+    let writeSpy: ReturnType<typeof vi.spyOn>;
+
+    beforeEach(() => {
+        writeSpy = vi.spyOn(process.stdout, "write").mockImplementation(() => true);
+    });
+
+    afterEach(() => {
+        writeSpy.mockRestore();
+    });
+
+    it("routes through the process writer synchronously", async () => {
+        expect.assertions(2);
+
+        writeCiOutput("hello from the default writer\n", "default");
+
+        expect(writeSpy).toHaveBeenCalledTimes(1);
+        expect(writeSpy).toHaveBeenCalledWith("hello from the default writer\n");
+
+        await flushCiOutput();
+    });
+});
diff --git a/packages/tooling/vis/src/tui/ci-output-writer.ts b/packages/tooling/vis/src/tui/ci-output-writer.ts
new file mode 100644
index 00000000..ecaade69
--- /dev/null
+++ b/packages/tooling/vis/src/tui/ci-output-writer.ts
@@ -0,0 +1,528 @@
+/**
+ * The minimal stream surface {@link createCiOutputWriter} needs. `process.stdout`
+ * and `process.stderr` satisfy it structurally, and tests build small fakes
+ * without touching the real process streams.
+ */
+export interface CiOutputStream {
+    /**
+     * Tear the stream down. Only ever called on a stream that has already
+     * stalled or errored — destroying it frees a wedged in-flight write so
+     * the writer (and the run behind it) can keep moving.
+     */
+    destroy?: () => unknown;
+    /** Remove a listener registered with {@link CiOutputStream.once}. */
+    off: (event: "close" | "drain" | "error", listener: () => void) => unknown;
+    /** Stream teardown notification. A closed stream is rerouted past, never crashed on. */
+    on: ((event: "close", listener: () => void) => unknown) & ((event: "error", listener: (error: Error) => void) => unknown);
+    /** One-shot wait helpers for `drain` / `error` / `close` while backpressured. */
+    once: (event: "close" | "drain" | "error", listener: () => void) => unknown;
+    /** Hand a chunk to the stream; `false` means the internal buffer is full (backpressure). */
+    write: (chunk: string) => boolean;
+}
+
+/**
+ * Which channel the writer currently hands output to.
+ *
+ * - `stdout` — the normal path.
+ * - `stderr` — taken over after stdout stalled or errored. stderr is a
+ *   separate pipe: when a CI runner wedges the stdout pipe, stderr usually
+ *   still flows, so the run keeps a visible log.
+ * - `dropped` — both pipes are unusable; output is discarded so the run can
+ *   still finish (and exit) instead of buffering into a void.
+ */
+export type CiOutputChannel = "stdout" | "stderr" | "dropped";
+
+export interface CiOutputWriterOptions {
+    /**
+     * Upper bound on a single `write()` call, in UTF-16 code units. Staying
+     * under the common 64 KiB pipe buffer means a healthy reader accepts
+     * every chunk synchronously, and a wedged reader can only ever swallow
+     * one bounded chunk.
+     */
+    chunkCodeUnits: number;
+
+    /** Clock injection for tests. Defaults to `Date.now`. */
+    now: () => number;
+
+    /**
+     * How long a stalled stderr write waits (measured on the stderr channel)
+     * before the writer drops output entirely. Only reachable when BOTH
+     * pipes are wedged; keeps the run completable instead of infinitely
+     * buffered.
+     */
+    stallDropMs: number;
+
+    /**
+     * How long a stalled stdout write waits before the writer gives up on
+     * stdout, destroys it to free the wedged write, and reroutes the
+     * remaining output to stderr. At most one chunk (≤
+     * {@link CiOutputWriterOptions.chunkCodeUnits}) is lost.
+     */
+    stallRerouteMs: number;
+
+    /**
+     * How long a stalled stdout write waits before a warning lands on
+     * stderr naming the pending output. The first sign of the failure mode
+     * this writer exists for: a CI runner that stopped draining our stdout.
+     */
+    stallWarnMs: number;
+}
+
+const DEFAULT_CI_OUTPUT_WRITER_OPTIONS: CiOutputWriterOptions = {
+    chunkCodeUnits: 32 * 1024,
+    now: Date.now,
+    stallDropMs: 30_000,
+    stallRerouteMs: 30_000,
+    stallWarnMs: 10_000,
+};
+
+/** One queued slice of output, with the label it arrived under (a task id). */
+interface PendingChunk {
+    label: string | undefined;
+    text: string;
+}
+
+type FlushSignal = "abandoned" | "close" | "drain" | "error";
+
+/**
+ * Split `text` into slices of at most `limit` UTF-16 code units.
+ *
+ * - Splits on a line boundary inside the window when one exists, so a reader
+ *   (and a log UI) always sees whole lines.
+ * - Never splits a surrogate pair: a boundary landing between a high and a
+ *   low surrogate is stepped back one unit, or the two halves would encode
+ *   as replacement characters and corrupt CJK/emoji output.
+ */
+export const splitCiOutputChunks = (text: string, limit: number): string[] => {
+    if (text.length <= limit) {
+        return [text];
+    }
+
+    const chunks: string[] = [];
+
+    let start = 0;
+
+    while (start < text.length) {
+        let end = Math.min(start + limit, text.length);
+
+        if (end < text.length) {
+            const lastNewline = text.lastIndexOf("\n", end - 1);
+
+            if (lastNewline >= start) {
+                end = lastNewline + 1;
+            }
+
+            // Step back off a surrogate pair: `end` must not sit between the
+            // high surrogate at `end - 1` and the low surrogate at `end`. A
+            // code point above the BMP range at `end - 1` means a complete
+            // pair starts there and spans the boundary.
+            if (end > start && (text.codePointAt(end - 1) ?? 0) > 0xff_ff) {
+                end -= 1;
+            }
+        }
+
+        chunks.push(text.slice(start, end));
+        start = end;
+    }
+
+    return chunks;
+};
+
+export interface CiOutputWriter {
+    /**
+     * Resolves once the queue is empty — immediately when nothing is
+     * pending, after the drain when backpressured, and right away once the
+     * channel is `dropped` (both pipes unusable) so no caller can hang on it.
+     */
+    flush: () => Promise<void>;
+    /** Diagnostic snapshot for tests and stall warnings. */
+    status: () => { channel: CiOutputChannel; pendingChunks: number; pendingCodeUnits: number };
+    /** Queue `text` for the active channel and (re)start the pump. Never blocks. */
+    write: (text: string, label?: string) => void;
+}
+
+/**
+ * Create a CI output writer over an explicit pair of streams.
+ *
+ * All CI-path output in `vis` flows through one writer per process so that
+ * a single wedged pipe cannot freeze (or silently swallow) the entire run
+ * log:
+ *
+ * - Writes are chunked to ≤ {@link CiOutputWriterOptions.chunkCodeUnits} and
+ *   honor backpressure: after a `write()` returns `false` the pump waits for
+ *   `drain` before committing the next chunk, so at most one chunk ever sits
+ *   inside the stream's internal buffer. The old single-giant-write path
+ *   could queue megabytes invisibly behind a stalled pipe and lose all of it
+ *   on `process.exit`.
+ * - The fast path stays synchronous: chunks keep being written while the
+ *   stream accepts them, so a healthy reader sees identical behavior to the
+ *   direct `process.stdout.write` calls this replaces.
+ * - A watchdog escalates a stuck channel: warn on stderr, then reroute to
+ *   stderr (destroying the wedged stdout to free its in-flight write), then
+ *   — only when stderr is wedged too — drop output so the run can finish.
+ *
+ * Observed incident (anolilab CI, 2026-10-20): a CI runner stopped draining
+ * `vis` stdout mid-dump of a completed task's ~256 KiB buffered output. The
+ * single unbounded write wedged, every later line queued invisibly behind
+ * it, the log froze mid-stack-trace, and the job burned an hour until it
+ * was cancelled — with every test that ran green.
+ */
+export const createCiOutputWriter = (stdout: CiOutputStream, stderr: CiOutputStream, options: Partial<CiOutputWriterOptions> = {}): CiOutputWriter => {
+    const config: CiOutputWriterOptions = { ...DEFAULT_CI_OUTPUT_WRITER_OPTIONS, ...options };
+    const tickMs = Math.min(250, Math.max(10, Math.floor(config.stallWarnMs / 4)));
+
+    const queue: PendingChunk[] = [];
+    const flushWaiters: (() => void)[] = [];
+    const errorListenersAttached = new Set<CiOutputStream>();
+
+    let channel: CiOutputChannel = "stdout";
+    let pumping = false;
+    let waiting = false;
+    let warned = false;
+    let lastProgressAt = 0;
+    let watchdogTimer: ReturnType<typeof setTimeout> | undefined;
+    let abandonCurrentWait: (() => void) | undefined;
+
+    const hasPendingWork = (): boolean => queue.length > 0 || waiting;
+
+    /**
+     * Stderr is the diagnostic channel — a wedged stdout must be reported
+     * on the OTHER pipe, and writing the report must never throw even if
+     * stderr itself is already gone.
+     */
+    const warnStderr = (message: string): void => {
+        try {
+            stderr.write(`${message}\n`);
+        } catch {
+            // stderr is gone too — nothing left to report on.
+        }
+    };
+
+    const resolveFlushWaiters = (): void => {
+        const waiters = flushWaiters.splice(0);
+
+        for (const waiter of waiters) {
+            waiter();
+        }
+    };
+
+    const scheduleWatchdog = (): void => {
+        if (channel === "dropped" || !hasPendingWork()) {
+            if (watchdogTimer) {
+                clearTimeout(watchdogTimer);
+                watchdogTimer = undefined;
+            }
+
+            return;
+        }
+
+        if (!watchdogTimer) {
+            watchdogTimer = setTimeout(tick, tickMs);
+            watchdogTimer.unref?.();
+        }
+    };
+
+    /** Escalation rung 2: stdout is wedged — free its write, move to stderr. */
+    const rerouteFromStdout = (stalledForMs: number): void => {
+        warned = true;
+
+        warnStderr(
+            `vis: stdout stalled for ${Math.round(stalledForMs / 1000)}s — dropping the wedged stdout write and rerouting the remaining output to stderr so the run can still finish; at most one ${config.chunkCodeUnits}-unit chunk of stdout output is lost.`,
+        );
+
+        // Destroying the stream frees the in-flight write that keeps the
+        // stream (and anything joining it) wedged. Verified against
+        // `process.stdout`: the pending write callback settles and the
+        // stream stops holding the run's log hostage.
+        try {
+            stdout.destroy?.();
+        } catch {
+            // An already-broken stream must never throw back into the watchdog.
+        }
+
+        channel = "stderr";
+        lastProgressAt = config.now();
+        abandonCurrentWait?.();
+    };
+
+    /** Escalation rung 3: both pipes are wedged — drop output, let the run finish. */
+    const dropFromStderr = (): void => {
+        try {
+            stderr.destroy?.();
+        } catch {
+            // An already-broken stream must never throw back into the watchdog.
+        }
+
+        channel = "dropped";
+        queue.length = 0;
+        lastProgressAt = config.now();
+        abandonCurrentWait?.();
+        resolveFlushWaiters();
+        scheduleWatchdog();
+    };
+
+    /**
+     * A stream errored (EPIPE and friends) or closed underneath us. The
+     * run's log must survive a dead pipe: reroute past stdout, drop past
+     * stderr. Never rethrow — an error listener that throws would take the
+     * process down.
+     */
+    const handleStreamError = (stream: CiOutputStream, error?: Error): void => {
+        if (stream === stdout && channel === "stdout") {
+            warned = true;
+
+            // `Error#code` exists on system errors (EPIPE, EDESTROYED …) but not
+            // on the base type; fall back to the message, then to a generic tag.
+            const code = (error as { code?: string } | undefined)?.code;
+
+            warnStderr(`vis: stdout writes failed (${code ?? error?.message ?? "stream closed or errored"}) — rerouting the remaining output to stderr.`);
+
+            channel = "stderr";
+            lastProgressAt = config.now();
+            abandonCurrentWait?.();
+
+            return;
+        }
+
+        if (stream === stderr && channel === "stderr") {
+            dropFromStderr();
+        }
+    };
+
+    const attachErrorListeners = (): void => {
+        if (!errorListenersAttached.has(stdout)) {
+            stdout.on("error", (error: Error) => {
+                handleStreamError(stdout, error);
+            });
+            stdout.on("close", () => {
+                handleStreamError(stdout);
+            });
+            errorListenersAttached.add(stdout);
+        }
+
+        if (!errorListenersAttached.has(stderr)) {
+            stderr.on("error", (error: Error) => {
+                handleStreamError(stderr, error);
+            });
+            stderr.on("close", () => {
+                handleStreamError(stderr);
+            });
+            errorListenersAttached.add(stderr);
+        }
+    };
+
+    /**
+     * Wait for the backpressured stream to make room (`drain`), die
+     * (`error` / `close`), or be switched away from by the watchdog
+     * (`abandoned`). Exactly one wait can be pending: the pump is single.
+     */
+    const waitForProgress = (stream: CiOutputStream): Promise<FlushSignal> =>
+        new Promise<FlushSignal>((resolve) => {
+            const settle = (signal: FlushSignal): void => {
+                stream.off("drain", onDrain);
+                stream.off("error", onError);
+                stream.off("close", onClose);
+
+                if (abandonCurrentWait === abandon) {
+                    abandonCurrentWait = undefined;
+                }
+
+                waiting = false;
+                resolve(signal);
+            };
+
+            const onDrain = (): void => {
+                settle("drain");
+            };
+            const onError = (): void => {
+                settle("error");
+            };
+            const onClose = (): void => {
+                settle("close");
+            };
+            const abandon = (): void => {
+                settle("abandoned");
+            };
+
+            waiting = true;
+            abandonCurrentWait = abandon;
+
+            stream.once("drain", onDrain);
+            stream.once("error", onError);
+            stream.once("close", onClose);
+        });
+
+    /**
+     * Commit queued chunks to the active stream. Runs synchronously while
+     * writes are accepted (identical behavior to direct `process.stdout.write`
+     * calls on a healthy reader) and only awaits once a write is
+     * backpressured — then at most one chunk is inside the stream's buffer.
+     */
+    const pump = async (): Promise<void> => {
+        if (pumping) {
+            return;
+        }
+
+        pumping = true;
+
+        try {
+            while (channel !== "dropped" && queue.length > 0) {
+                const stream = channel === "stderr" ? stderr : stdout;
+                const chunk = queue[0]!;
+
+                // Commit the chunk before checking acceptance: `write()`
+                // returning `false` still hands the chunk over — it already
+                // sits in the stream's buffer and must never be re-sent.
+                const accepted = stream.write(chunk.text);
+
+                queue.shift();
+
+                if (accepted) {
+                    lastProgressAt = config.now();
+                    warned = false;
+
+                    continue;
+                }
+
+                const signal = await waitForProgress(stream);
+
+                if (signal === "drain") {
+                    lastProgressAt = config.now();
+                    warned = false;
+                }
+
+                // "error" / "close": the error/close listeners attached in
+                // `attachErrorListeners` own the channel transition — the
+                // loop just re-reads `channel` and continues.
+                // "abandoned": the watchdog switched channels the same way.
+            }
+        } catch (error) {
+            // A `write()` that throws synchronously (a broken stream) is the
+            // same failure as an `error` event: reroute or drop, never let
+            // the fire-and-forget pump become an unhandled rejection.
+            handleStreamError(channel === "stderr" ? stderr : stdout, error instanceof Error ? error : undefined);
+        } finally {
+            pumping = false;
+            scheduleWatchdog();
+            resolveFlushWaiters();
+        }
+    };
+
+    const tick = (): void => {
+        watchdogTimer = undefined;
+
+        if (channel === "dropped" || !hasPendingWork()) {
+            resolveFlushWaiters();
+
+            return;
+        }
+
+        const stalledForMs = config.now() - lastProgressAt;
+
+        if (channel === "stdout") {
+            if (stalledForMs >= config.stallRerouteMs) {
+                rerouteFromStdout(stalledForMs);
+            } else if (stalledForMs >= config.stallWarnMs && !warned) {
+                warned = true;
+
+                const { pendingChunks, pendingCodeUnits } = writerStatus();
+                const label = queue.at(0)?.label;
+
+                warnStderr(
+                    `vis: stdout writes stalled for ${Math.round(stalledForMs / 1000)}s (${pendingChunks} chunk(s), ~${Math.round(pendingCodeUnits / 1024)} KB pending${label ? ` for "${label}"` : ""}); waiting for the log reader to drain.`,
+                );
+            }
+        } else if (stalledForMs >= config.stallDropMs) {
+            dropFromStderr();
+        }
+
+        scheduleWatchdog();
+
+        // `queue.length` alone is the guard: the drop path clears the queue,
+        // and the reroute path leaves it (the pump then continues on stderr).
+        if (queue.length > 0) {
+            // eslint-disable-next-line no-void -- void marks the pump promise as intentionally fire-and-forget
+            void pump();
+        }
+    };
+
+    const writerStatus = (): { channel: CiOutputChannel; pendingChunks: number; pendingCodeUnits: number } => {
+        return {
+            channel,
+            pendingChunks: queue.length,
+            pendingCodeUnits: queue.reduce((total, chunk) => total + chunk.text.length, 0),
+        };
+    };
+
+    const write = (text: string, label?: string): void => {
+        if (channel === "dropped" || text.length === 0) {
+            return;
+        }
+
+        // A fresh busy episode resets the stall clock so the reader gets a
+        // full window; continuous new work must NOT reset it, or a chatty
+        // wedged pipe would never cross the escalation threshold.
+        if (!hasPendingWork()) {
+            lastProgressAt = config.now();
+        }
+
+        for (const textChunk of splitCiOutputChunks(text, config.chunkCodeUnits)) {
+            queue.push({ label, text: textChunk });
+        }
+
+        attachErrorListeners();
+
+        // eslint-disable-next-line no-void -- void marks the pump promise as intentionally fire-and-forget
+        void pump();
+        scheduleWatchdog();
+    };
+
+    const flush = (): Promise<void> =>
+        new Promise<void>((resolve) => {
+            if (channel === "dropped" || !hasPendingWork()) {
+                resolve();
+
+                return;
+            }
+
+            flushWaiters.push(resolve);
+            scheduleWatchdog();
+        });
+
+    return {
+        flush,
+        status: writerStatus,
+        write,
+    };
+};
+
+let defaultCiOutputWriter: CiOutputWriter | undefined;
+
+/**
+ * The process-wide writer over the real `process.stdout` / `process.stderr`.
+ * Lazily created so tests that spy on `process.stdout.write` (the existing
+ * tui suites) install their spy before any chunk is committed.
+ */
+const getDefaultCiOutputWriter = (): CiOutputWriter => {
+    defaultCiOutputWriter ??= createCiOutputWriter(process.stdout, process.stderr);
+
+    return defaultCiOutputWriter;
+};
+
+/**
+ * Queue a line of CI-path output on the process writer. Drop-in for a direct
+ * `process.stdout.write` call: synchronous on a healthy reader, never
+ * blocking when the pipe stalls, and rerouted to stderr when it wedges.
+ *
+ * `label` names the logical unit the text belongs to (a task id, the run
+ * summary) so a stall warning can say WHAT is stuck, not just how much.
+ */
+export const writeCiOutput = (text: string, label?: string): void => {
+    getDefaultCiOutputWriter().write(text, label);
+};
+
+/**
+ * Await the process writer's queue. Exported for tests and for any exit path
+ * that wants to confirm the log landed before finishing.
+ */
+export const flushCiOutput = (): Promise<void> => getDefaultCiOutputWriter().flush();
diff --git a/packages/tooling/vis/src/tui/static-life-cycle.ts b/packages/tooling/vis/src/tui/static-life-cycle.ts
index a24b82e6..dcb93491 100644
--- a/packages/tooling/vis/src/tui/static-life-cycle.ts
+++ b/packages/tooling/vis/src/tui/static-life-cycle.ts
@@ -4,6 +4,7 @@ import { Text } from "@visulima/tui/components/text";
 import React from "react";
 
 import type { VisTargetOptions } from "../task/target-options";
+import { writeCiOutput } from "./ci-output-writer";
 import CommandSummary from "./components/command-summary";
 import Header from "./components/header";
 import { renderFailureOutput } from "./failure-render";
@@ -92,6 +93,11 @@ export const declaredFallbackMessage = (task: Task): string => `${task.id}: trac
 /**
  * A lifecycle handler for CI environments that produces static, append-only output.
  * No cursor manipulation — just linear log lines.
+ *
+ * Every line goes through {@link writeCiOutput} (see `ci-output-writer.ts`)
+ * so CI-path output shares one bounded, backpressure-aware path — a runner
+ * that stops draining our stdout pipe cannot freeze or silently swallow
+ * the rest of the run's log.
  */
 export class StaticOutputLifeCycle implements LifeCycleInterface {
     readonly #projectNames: string[];
@@ -133,21 +139,21 @@ export class StaticOutputLifeCycle implements LifeCycleInterface {
         const title = `Running ${formatTargetsAndProjects(this.#projectNames, this.#targets, this.#tasks)}`;
         const header = renderToString(React.createElement(Header, { title, variant: "info" }), { columns });
 
-        process.stdout.write(header);
+        writeCiOutput(header, "run header");
 
         // Print overrides if any (skip internal "command" key)
         const firstTask = this.#tasks[0];
         const overrideEntries = firstTask?.overrides ? Object.entries(firstTask.overrides).filter(([flag]) => flag !== "command") : [];
 
         if (overrideEntries.length > 0) {
-            process.stdout.write(`\n  With additional flags:\n`);
+            writeCiOutput(`\n  With additional flags:\n`, "run header");
 
             for (const [flag, value] of overrideEntries) {
-                process.stdout.write(`${formatFlags("    ", flag, value)}\n`);
+                writeCiOutput(`${formatFlags("    ", flag, value)}\n`, "run header");
             }
         }
 
-        process.stdout.write("\n");
+        writeCiOutput("\n", "run header");
     }
 
     // fallow-ignore-next-line unused-class-member -- public TUI store/life-cycle method driven by the React TUI components
@@ -157,7 +163,7 @@ export class StaticOutputLifeCycle implements LifeCycleInterface {
         for (const task of tasks) {
             const line = renderToString(React.createElement(Text, null, React.createElement(Text, { dimColor: true }, ">"), ` ${task.id}`), { columns });
 
-            process.stdout.write(`${line}\n`);
+            writeCiOutput(`${line}\n`, task.id);
         }
     }
 
@@ -198,7 +204,7 @@ export class StaticOutputLifeCycle implements LifeCycleInterface {
                 { columns },
             );
 
-            process.stdout.write(`${line}\n`);
+            writeCiOutput(`${line}\n`, result.task.id);
         }
     }
 
@@ -206,7 +212,7 @@ export class StaticOutputLifeCycle implements LifeCycleInterface {
         const columns = process.stdout.columns || 80;
         const line = renderToString(React.createElement(Text, { dimColor: true }, `  ⓘ ${message}`), { columns });
 
-        process.stdout.write(`${line}\n`);
+        writeCiOutput(`${line}\n`, "cache notice");
     }
 
     // fallow-ignore-next-line unused-class-member -- public TUI store/life-cycle method driven by the React TUI components
@@ -267,7 +273,7 @@ export class StaticOutputLifeCycle implements LifeCycleInterface {
         const skippedIds = this.#tasks.filter((t) => !this.#allCompletedTasks.has(t.id)).map((t) => t.id);
 
         // Empty line between task output and summary
-        process.stdout.write("\n");
+        writeCiOutput("\n", "run summary");
 
         const columns = process.stdout.columns || 80;
         const summary = renderToString(
@@ -286,6 +292,6 @@ export class StaticOutputLifeCycle implements LifeCycleInterface {
             { columns },
         );
 
-        process.stdout.write(`${summary}\n`);
+        writeCiOutput(`${summary}\n`, "run summary");
     }
 }
diff --git a/packages/tooling/vis/src/tui/status-utils.ts b/packages/tooling/vis/src/tui/status-utils.ts
index 3262523e..3d5b951c 100644
--- a/packages/tooling/vis/src/tui/status-utils.ts
+++ b/packages/tooling/vis/src/tui/status-utils.ts
@@ -5,6 +5,7 @@ import { renderToString } from "@visulima/tui";
 import { Text } from "@visulima/tui/components/text";
 import React from "react";
 
+import { writeCiOutput } from "./ci-output-writer";
 import { CROSS, DASH, TICK } from "./symbols";
 
 export const isCacheStatus = (status: string): boolean => status === "local-cache" || status === "local-cache-kept-existing" || status === "remote-cache";
@@ -148,6 +149,12 @@ const toGitLabSectionKey = (taskId: string): string => {
  * intentionally left expanded so the failure is visible without an
  * extra click. Pass `mode` from `vis-config.ts → run.ciGrouping`;
  * `auto` (the default) detects the runner via env vars.
+ *
+ * All writes flow through {@link writeCiOutput} (see `ci-output-writer.ts`)
+ * so a CI runner that stops draining our stdout pipe cannot wedge the
+ * whole log behind one unbounded write: chunks stay bounded, a stall is
+ * reported on stderr, and the remaining output is rerouted rather than
+ * silently swallowed.
  */
 export const logCommandOutputCI = (taskId: string, status: TaskStatus, output: string, mode: CiGroupingMode | undefined = "auto"): void => {
     const trimmed = output.trim();
@@ -166,9 +173,9 @@ export const logCommandOutputCI = (taskId: string, status: TaskStatus, output: s
     // otherwise tidy log; the failed task's group sits at the tail of the log
     // and is a single click away.
     if (grouping === "github") {
-        process.stdout.write(`::group::${getStatusIcon(status)} ${taskId}${EOL}`);
-        process.stdout.write(trimmed + EOL);
-        process.stdout.write(`::endgroup::${EOL}`);
+        writeCiOutput(`::group::${getStatusIcon(status)} ${taskId}${EOL}`, taskId);
+        writeCiOutput(trimmed + EOL, taskId);
+        writeCiOutput(`::endgroup::${EOL}`, taskId);
 
         return;
     }
@@ -188,12 +195,12 @@ export const logCommandOutputCI = (taskId: string, status: TaskStatus, output: s
         // bytes from the rendered log so the directive itself stays hidden.
         const eraseLine = "";
 
-        process.stdout.write(`${eraseLine}section_start:${String(startTs)}:${key}${collapsed}\r${eraseLine}${getStatusIcon(status)} ${taskId}${EOL}`);
-        process.stdout.write(trimmed + EOL);
+        writeCiOutput(`${eraseLine}section_start:${String(startTs)}:${key}${collapsed}\r${eraseLine}${getStatusIcon(status)} ${taskId}${EOL}`, taskId);
+        writeCiOutput(trimmed + EOL, taskId);
 
         const endTs = Math.floor(Date.now() / 1000);
 
-        process.stdout.write(`${eraseLine}section_end:${String(endTs)}:${key}\r${eraseLine}${EOL}`);
+        writeCiOutput(`${eraseLine}section_end:${String(endTs)}:${key}\r${eraseLine}${EOL}`, taskId);
 
         return;
     }
@@ -205,8 +212,8 @@ export const logCommandOutputCI = (taskId: string, status: TaskStatus, output: s
         // a failed task is expanded by default while successes stay folded.
         const heading = status === "failure" ? "+++" : "---";
 
-        process.stdout.write(`${heading} ${getStatusIcon(status)} ${taskId}${EOL}`);
-        process.stdout.write(trimmed + EOL);
+        writeCiOutput(`${heading} ${getStatusIcon(status)} ${taskId}${EOL}`, taskId);
+        writeCiOutput(trimmed + EOL, taskId);
 
         return;
     }
@@ -215,9 +222,9 @@ export const logCommandOutputCI = (taskId: string, status: TaskStatus, output: s
         // Azure Pipelines only offers a collapsed `##[group]` — there is no
         // expanded-group directive — so, like GitHub, a failed task is
         // grouped collapsed and sits one click from the error.
-        process.stdout.write(`##[group]${getStatusIcon(status)} ${taskId}${EOL}`);
-        process.stdout.write(trimmed + EOL);
-        process.stdout.write(`##[endgroup]${EOL}`);
+        writeCiOutput(`##[group]${getStatusIcon(status)} ${taskId}${EOL}`, taskId);
+        writeCiOutput(trimmed + EOL, taskId);
+        writeCiOutput(`##[endgroup]${EOL}`, taskId);
 
         return;
     }
@@ -228,8 +235,8 @@ export const logCommandOutputCI = (taskId: string, status: TaskStatus, output: s
     const prefix = getStatusPrefix(status);
     const boldTaskId = renderToString(React.createElement(Text, { bold: true }, taskId), { columns: width }).trim();
 
-    process.stdout.write(`${separator}${EOL}`);
-    process.stdout.write(`${prefix} ${boldTaskId}${EOL}`);
-    process.stdout.write(trimmed + EOL);
-    process.stdout.write(`${separator}${EOL}`);
+    writeCiOutput(`${separator}${EOL}`, taskId);
+    writeCiOutput(`${prefix} ${boldTaskId}${EOL}`, taskId);
+    writeCiOutput(trimmed + EOL, taskId);
+    writeCiOutput(`${separator}${EOL}`, taskId);
 };
diff --git a/packages/tooling/vis/src/tui/summary-life-cycle.ts b/packages/tooling/vis/src/tui/summary-life-cycle.ts
index ab4af7aa..49b1c75e 100644
--- a/packages/tooling/vis/src/tui/summary-life-cycle.ts
+++ b/packages/tooling/vis/src/tui/summary-life-cycle.ts
@@ -2,6 +2,7 @@ import type { LifeCycleInterface, Task, TaskResult, TaskStatus } from "@visulima
 import { renderToString } from "@visulima/tui";
 import React from "react";
 
+import { writeCiOutput } from "./ci-output-writer";
 import TaskSummaryView from "./components/task-summary-view";
 
 interface TaskSummaryEntry {
@@ -77,7 +78,7 @@ export class SummaryLifeCycle implements LifeCycleInterface {
         const columns = process.stdout.columns || 80;
         const output = renderToString(React.createElement(TaskSummaryView, { entries }), { columns });
 
-        process.stdout.write(`\n${output}\n`);
+        writeCiOutput(`\n${output}\n`, "run summary");
     }
 }
 

From 9c56827112f7e4d07b42c16bf503a3a2ce9ae93a Mon Sep 17 00:00:00 2001
From: Daniel Bannert <d.bannert@anolilab.de>
Date: Tue, 20 Oct 2026 13:46:51 +0200
Subject: [PATCH 2/2] ci: disable the signatures audit in ci.yml like every
 other workflow
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The CI workflow was the only one not passing run-signatures-audit: false
to the node setup step, so it inherited the default and ran `npm audit
signatures` — which fails with ETARGET on the never-published workspace
binding placeholder `@visulima/secret-scanner-binding-*@0.0.0-alpha.0`
that pnpm installs from the workspace. main has been red on this since
2026-10-19; build-native, codspeed, coverage-baseline, data-sync, lint,
semantic-release and preview-release all already disable it.
---
 .github/workflows/ci.yml | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 571e9521..995b1aed 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -37,6 +37,13 @@ jobs:
                   cache-prefix: "vis"
                   skip-playwright: "true"
                   upgrade-npm-version: false
+                  # TODO: re-enable once pnpm/npm audit works with the unpublished workspace
+                  # binding packages — `@visulima/secret-scanner-binding-*` installs as the
+                  # placeholder `0.0.0-alpha.0`, which `npm audit signatures` looks up in the
+                  # registry and fails on with ETARGET. Every other workflow disables it the
+                  # same way (see build-native.yml, codspeed.yml, lint.yml).
+                  run-signatures-audit: false
+                  run-npm-audit: false
 
             - name: "Download native build artifacts"
               uses: "./.github/actions/native-test-artifacts"