Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 29 additions & 3 deletions docs/users/features/dual-output.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -223,6 +223,28 @@ 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)

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.

```bash
mkfifo /tmp/qwen-events.jsonl /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
Expand Down Expand Up @@ -325,6 +347,10 @@ 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, the bridge buffers events internally. If the buffer exceeds
1 MB (no consumer is draining), the bridge disables itself. 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.
Expand Down
63 changes: 63 additions & 0 deletions packages/cli/src/dualOutput/DualOutputBridge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -199,4 +200,66 @@ describe('DualOutputBridge', () => {
expect(() => bridge!.emitControlResponse('req', true)).not.toThrow();
});
});

describe('FIFO (named pipe) support', () => {
let fifoPath: string;

beforeEach(() => {
fifoPath = path.join(tmpDir, 'events.fifo');
try {
execSync(`mkfifo "${fifoPath}"`);
} catch {
// mkfifo not available (Windows) — skip these tests
}
});

it('does not block when opened without a reader connected', () => {
if (!fs.existsSync(fifoPath) || !fs.statSync(fifoPath).isFIFO()) {
Comment thread
chiga0 marked this conversation as resolved.
Resolved
return; // skip on platforms without mkfifo
}

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 () => {
if (!fs.existsSync(fifoPath) || !fs.statSync(fifoPath).isFIFO()) {
return; // skip on platforms without mkfifo
}

bridge = new DualOutputBridge(config, { filePath: fifoPath });
bridge.emitSystemMessage('test_event', { key: 'value' });

// Connect a reader after writes
const received = await new Promise<string>((resolve) => {
const chunks: Buffer[] = [];
const reader = fs.createReadStream(fifoPath);
reader.on('data', (chunk) => chunks.push(chunk as Buffer));
// Close the bridge to flush + EOF
bridge!.shutdown().then(() => {
reader.on('end', () => resolve(Buffer.concat(chunks).toString()));
});
});

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' },
});
});
});
});
35 changes: 32 additions & 3 deletions packages/cli/src/dualOutput/DualOutputBridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
*/
Expand Down Expand Up @@ -117,9 +124,19 @@ 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.
const fd = openSync(
target.filePath,
constants.O_RDWR | constants.O_NONBLOCK,
);
this.stream = createWriteStream('', { fd });
} else if (code === 'ENOENT') {
// Regular file doesn't exist yet — create it.
this.stream = createWriteStream(target.filePath, { flags: 'w' });
} else {
throw err;
Expand Down Expand Up @@ -165,6 +182,7 @@ export class DualOutputBridge {

processEvent(event: ServerGeminiStreamEvent): void {
if (!this.active) return;
if (this.isBufferOverflowing()) return;
Comment thread
chiga0 marked this conversation as resolved.
Resolved
try {
this.adapter.processEvent(event);
} catch (err) {
Expand Down Expand Up @@ -221,6 +239,17 @@ export class DualOutputBridge {
return this.active;
}

private isBufferOverflowing(): boolean {
if (this.stream.writableLength > MAX_BUFFERED_BYTES) {
debugLogger.warn(
'DualOutput: buffered data exceeds limit, disabling (no consumer draining?)',
);
this.active = false;
return true;
Comment thread
chiga0 marked this conversation as resolved.
Resolved
}
Comment thread
chiga0 marked this conversation as resolved.
Resolved
return false;
}

/**
* Emits a `can_use_tool` permission request so an external consumer can
* approve or deny the tool call. Pairs with {@link emitControlResponse}.
Expand Down
Loading