fix(dual-output): prevent FIFO blocking on startup when no reader connected

docs/users/features/dual-output.md

@@ -199,10 +199,10 @@ wrappers that throw away stdout anyway.
 
 ## Quick start
 
-Run Qwen Code with all three channels enabled:
+Run Qwen Code with both channels enabled using regular files:
 
 ```bash
-mkfifo /tmp/qwen-events.jsonl /tmp/qwen-input.jsonl
+touch /tmp/qwen-events.jsonl /tmp/qwen-input.jsonl
 qwen \
   --json-file /tmp/qwen-events.jsonl \
   --input-file /tmp/qwen-input.jsonl
@@ -211,7 +211,7 @@ qwen \
 In a second terminal, tail the event stream:
 
 ```bash
-cat /tmp/qwen-events.jsonl
+tail -f /tmp/qwen-events.jsonl
 ```
 
 In a third terminal, push a prompt into the running TUI:
@@ -223,6 +223,33 @@ echo '{"type":"submit","text":"Explain this repo"}' >> /tmp/qwen-input.jsonl
 The prompt appears in the TUI exactly as if the user typed it, and the
 streaming response is mirrored on `/tmp/qwen-events.jsonl`.
 
+### Using FIFOs (named pipes) for event output
+
+FIFOs deliver lower latency than regular files (no disk I/O) and work
+well when both sides are on the same host. The bridge opens FIFOs with
+`O_RDWR | O_NONBLOCK`, so it **does not block** even if no reader is
+connected yet — events are buffered in the kernel pipe buffer until a
+reader attaches.
+
+> **Note:** `--input-file` requires a regular file (not a FIFO) because
+> the watcher relies on `stat.size` to detect new data, which is always
+> 0 for FIFOs.
+
+```bash
+mkfifo /tmp/qwen-events.jsonl
+touch /tmp/qwen-input.jsonl
+qwen \
+  --json-file /tmp/qwen-events.jsonl \
+  --input-file /tmp/qwen-input.jsonl
+# TUI starts immediately — no need to start a reader first.
+
+# In a second terminal, connect whenever ready:
+cat /tmp/qwen-events.jsonl
+```
+
+If no reader ever connects, the bridge auto-disables once the internal
+buffer exceeds 1 MB. The TUI continues running normally.
+
 ## Output event schema
 
 Events are emitted as JSON Lines (one object per line). The schema is the same
@@ -325,6 +352,13 @@ polling — events are written synchronously as the TUI emits them.
 - **Consumer disconnect.** If the reader on the other side of the channel
   goes away (`EPIPE`), the bridge silently disables itself and the TUI
   keeps running. No retry.
+- **FIFO buffer overflow.** When writing to a FIFO with no reader
+  attached, events buffer in the kernel pipe (~64 KB on Linux) and the
+  Node.js WriteStream. Once the pipe is full or the internal buffer
+  exceeds 1 MB, the bridge disables itself and closes the fd. No
+  `session_end` is emitted in this case — consumers should treat a
+  closed stream without `session_end` as an abnormal termination. The
+  TUI continues running normally.
 - **Adapter exception.** Any exception thrown while emitting an event is
   caught, logged, and disables the bridge. The TUI is never crashed by a
   dual-output failure.

packages/cli/src/dualOutput/DualOutputBridge.test.ts

@@ -8,6 +8,7 @@ import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
 import * as fs from 'node:fs';
 import * as os from 'node:os';
 import * as path from 'node:path';
+import { execSync } from 'node:child_process';
 import type { Config } from '@qwen-code/qwen-code-core';
 import {
   DualOutputBridge,
@@ -199,4 +200,119 @@ describe('DualOutputBridge', () => {
       expect(() => bridge!.emitControlResponse('req', true)).not.toThrow();
     });
   });
+
+  describe('buffer overflow guard', () => {
+    it('disables itself when buffered data exceeds 1 MB', () => {
+      bridge = new DualOutputBridge(config, { filePath: target });
+      expect(bridge.isConnected).toBe(true);
+
+      // Simulate a bloated buffer by overriding writableLength
+      Object.defineProperty(bridge['stream'], 'writableLength', {
+        value: 1024 * 1024 + 1,
+      });
+
+      // Any write method should trigger the guard
+      bridge.emitSystemMessage('test', {});
+      expect(bridge.isConnected).toBe(false);
+    });
+
+    it('destroys the stream on overflow so consumers receive EOF', () => {
+      bridge = new DualOutputBridge(config, { filePath: target });
+      const destroySpy = vi.spyOn(bridge['stream'], 'destroy');
+
+      Object.defineProperty(bridge['stream'], 'writableLength', {
+        value: 1024 * 1024 + 1,
+      });
+
+      bridge.emitSystemMessage('test', {});
+      expect(destroySpy).toHaveBeenCalled();
+    });
+
+    it('shutdown resolves immediately after buffer overflow destroys stream', async () => {
+      bridge = new DualOutputBridge(config, { filePath: target });
+
+      Object.defineProperty(bridge['stream'], 'writableLength', {
+        value: 1024 * 1024 + 1,
+      });
+      bridge.emitSystemMessage('test', {});
+      expect(bridge.isConnected).toBe(false);
+
+      await expect(bridge.shutdown()).resolves.toBeUndefined();
+    });
+
+    it('disables on ERR_SYSTEM_ERROR stream error', () => {
+      bridge = new DualOutputBridge(config, { filePath: target });
+      expect(bridge.isConnected).toBe(true);
+
+      bridge['stream'].emit(
+        'error',
+        Object.assign(new Error('EAGAIN'), { code: 'ERR_SYSTEM_ERROR' }),
+      );
+      expect(bridge.isConnected).toBe(false);
+    });
+  });
+
+  describe.skipIf(process.platform === 'win32' || process.getuid?.() === 0)(
+    'FIFO (named pipe) support',
+    () => {
+      let fifoPath: string;
+
+      beforeEach(() => {
+        fifoPath = path.join(tmpDir, 'events.fifo');
+        execSync(`mkfifo "${fifoPath}"`);
+      });
+
+      it('does not block when opened without a reader connected', () => {
+        const start = Date.now();
+        bridge = new DualOutputBridge(config, { filePath: fifoPath });
+        const elapsed = Date.now() - start;
+
+        expect(elapsed).toBeLessThan(500);
+        expect(bridge.isConnected).toBe(true);
+      });
+
+      it('delivers events to a reader that connects after construction', async () => {
+        bridge = new DualOutputBridge(config, { filePath: fifoPath });
+        bridge.emitSystemMessage('test_event', { key: 'value' });
+
+        const received = await new Promise<string>((resolve) => {
+          const chunks: Buffer[] = [];
+          const reader = fs.createReadStream(fifoPath);
+          reader.on('data', (chunk) => chunks.push(chunk as Buffer));
+          reader.on('end', () => resolve(Buffer.concat(chunks).toString()));
+          reader.on('open', () => bridge!.shutdown());
+        });
+
+        const lines = received
+          .split('\n')
+          .filter(Boolean)
+          .map((l) => JSON.parse(l));
+        expect(lines[0]).toMatchObject({
+          type: 'system',
+          subtype: 'session_start',
+        });
+        const testEvent = lines.find(
+          (l: Record<string, unknown>) =>
+            l['type'] === 'system' && l['subtype'] === 'test_event',
+        );
+        expect(testEvent).toMatchObject({
+          data: { key: 'value' },
+        });
+      });
+
+      it('throws actionable error when FIFO lacks read permission', () => {
+        const noReadFifo = path.join(tmpDir, 'no-read.fifo');
+        // chmod 0200 (write-only): first openSync(O_WRONLY) returns ENXIO
+        // (no reader), retry with O_RDWR fails EACCES (no read permission)
+        execSync(`mkfifo "${noReadFifo}" && chmod 0200 "${noReadFifo}"`);
+        try {
+          expect(
+            () => new DualOutputBridge(config, { filePath: noReadFifo }),
+          ).toThrow(/permission denied opening FIFO for read-write/);
+        } finally {
+          execSync(`chmod 644 "${noReadFifo}"`);
+        }
+      });
+    },
+  );
 });

packages/cli/src/dualOutput/DualOutputBridge.ts

@@ -52,6 +52,13 @@ export const SUPPORTED_EVENTS = [
  */
 export const DUAL_OUTPUT_PROTOCOL_VERSION = 1;
 
+/**
+ * Maximum bytes buffered in the Node.js WriteStream before the bridge
+ * self-disables. Guards against unbounded memory growth when the output
+ * target is a FIFO opened with O_RDWR (no EPIPE on reader disconnect).
+ */
+const MAX_BUFFERED_BYTES = 1024 * 1024; // 1 MB
+
 /**
  * Optional metadata wired into the `session_start` capability handshake.
  */
@@ -107,8 +114,8 @@ export class DualOutputBridge {
     } else {
       // Open with O_WRONLY|O_NONBLOCK to avoid blocking the event loop on FIFOs.
       // On FIFO, a regular open(O_WRONLY) blocks until a reader connects.
-      // O_NONBLOCK makes it return immediately (ENXIO if no reader yet, which
-      // createWriteStream handles via its internal retry/error mechanism).
+      // O_NONBLOCK makes openSync return immediately; if no reader is
+      // connected yet (ENXIO), the catch block below retries with O_RDWR.
       try {
         const fd = openSync(
           target.filePath,
@@ -117,9 +124,30 @@ export class DualOutputBridge {
         this.stream = createWriteStream('', { fd });
       } catch (err) {
         const code = (err as NodeJS.ErrnoException).code;
-        // ENXIO: FIFO has no reader yet — fall back to blocking open.
-        // ENOENT: regular file doesn't exist yet — create it.
-        if (code === 'ENXIO' || code === 'ENOENT') {
+        if (code === 'ENXIO') {
+          // FIFO with no reader connected yet. Use O_RDWR | O_NONBLOCK so
+          // the open returns immediately (POSIX: process is both reader and
+          // writer, satisfying the "at least one reader" requirement).
+          // Trade-off: EPIPE won't fire on reader disconnect; the bridge
+          // self-disables when the pipe buffer fills instead.
+          try {
+            const fd = openSync(
+              target.filePath,
+              constants.O_RDWR | constants.O_NONBLOCK,
+            );
+            this.stream = createWriteStream('', { fd });
+          } catch (retryErr) {
+            if ((retryErr as NodeJS.ErrnoException).code === 'EACCES') {
+              throw new Error(
+                `--json-file "${target.filePath}": permission denied opening FIFO for read-write. ` +
+                  'Check read/write permissions on the file and its parent directories, ' +
+                  'or start a reader before launching Qwen Code.',
+              );
+            }
+            throw retryErr;
+          }
+        } else if (code === 'ENOENT') {
+          // Regular file doesn't exist yet — create it.
           this.stream = createWriteStream(target.filePath, { flags: 'w' });
         } else {
           throw err;
@@ -129,9 +157,13 @@ export class DualOutputBridge {
 
     this.stream.on('error', (err) => {
       const code = (err as NodeJS.ErrnoException).code;
-      // Consumer disconnected — gracefully stop writing, don't crash the TUI
       if (code === 'EPIPE' || code === 'ERR_STREAM_DESTROYED') {
         debugLogger.warn('DualOutput: consumer disconnected, disabling');
+      } else if (code === 'ERR_SYSTEM_ERROR') {
+        debugLogger.warn(
+          'DualOutput: system error on stream, disabling:',
+          (err as NodeJS.ErrnoException).message,
+        );
       } else {
         debugLogger.error('DualOutput stream error:', err);
       }
@@ -164,6 +196,8 @@ export class DualOutputBridge {
   }
 
   processEvent(event: ServerGeminiStreamEvent): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.processEvent(event);
@@ -174,6 +208,8 @@ export class DualOutputBridge {
   }
 
   startAssistantMessage(): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.startAssistantMessage();
@@ -184,6 +220,8 @@ export class DualOutputBridge {
   }
 
   finalizeAssistantMessage(): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.finalizeAssistantMessage();
@@ -194,6 +232,8 @@ export class DualOutputBridge {
   }
 
   emitUserMessage(parts: Part[]): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.emitUserMessage(parts);
@@ -207,6 +247,8 @@ export class DualOutputBridge {
     request: ToolCallRequestInfo,
     response: ToolCallResponseInfo,
   ): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.emitToolResult(request, response);
@@ -221,6 +263,16 @@ export class DualOutputBridge {
     return this.active;
   }
 
+  private disableIfBufferOverflowed(): void {
+    if (this.stream.writableLength > MAX_BUFFERED_BYTES) {
+      debugLogger.warn(
+        'DualOutput: buffered data exceeds limit, disabling (no consumer draining?)',
+      );
+      this.active = false;
+      this.stream.destroy();
+    }
+  }
+
   /**
    * Emits a `can_use_tool` permission request so an external consumer can
    * approve or deny the tool call. Pairs with {@link emitControlResponse}.
@@ -232,6 +284,8 @@ export class DualOutputBridge {
     input: unknown,
     blockedPath: string | null = null,
   ): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.emitPermissionRequest(
@@ -252,6 +306,8 @@ export class DualOutputBridge {
    * the external consumer) so all observers stay in sync.
    */
   emitControlResponse(requestId: string, allowed: boolean): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.emitControlResponse(requestId, allowed);
@@ -268,6 +324,8 @@ export class DualOutputBridge {
    * consumers retry or surface the error instead of silently hanging.
    */
   emitControlError(requestId: string, message: string): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.emitControlError(requestId, message);
@@ -279,6 +337,8 @@ export class DualOutputBridge {
 
   /** General-purpose system event escape hatch. */
   emitSystemMessage(subtype: string, data?: unknown): void {
+    if (!this.active) return;
+    this.disableIfBufferOverflowed();
     if (!this.active) return;
     try {
       this.adapter.emitSystemMessage(subtype, data);
@@ -305,7 +365,7 @@ export class DualOutputBridge {
     }
     this.active = false;
     this.shutdownPromise = new Promise((resolve) => {
-      if (this.stream.closed) {
+      if (this.stream.closed || this.stream.destroyed) {
         resolve();
         return;
       }