basant307/AI_Governance_Project

1/**2 * @license3 * Copyright 2025 Qwen Team4 * SPDX-License-Identifier: Apache-2.05 */6 7import {8  createWriteStream,9  fstatSync,10  openSync,11  constants,12  type WriteStream,13} from 'node:fs';14import type {15  Config,16  ServerGeminiStreamEvent,17  ToolCallRequestInfo,18  ToolCallResponseInfo,19} from '@qwen-code/qwen-code-core';20import { createDebugLogger } from '@qwen-code/qwen-code-core';21import type { Part } from '@google/genai';22import { StreamJsonOutputAdapter } from '../nonInteractive/io/index.js';23 24const debugLogger = createDebugLogger('DUAL_OUTPUT');25 26/**27 * Structured-event kinds this bridge version is known to emit. Exposed to28 * consumers in `session_start.data.supported_events` so they can29 * feature-detect rather than sniffing the stream or hard-coding a minimum30 * CLI version.31 *32 * When adding a new event kind, append it here and bump the handshake33 * `protocol_version` below so consumers can gate on the combination.34 */35export const SUPPORTED_EVENTS = [36  'system',37  'user',38  'assistant',39  'stream_event',40  'result',41  'control_request',42  'control_response',43] as const;44 45/**46 * Monotonically-increasing integer bumped whenever the wire protocol47 * changes in a way consumers might care about (new event types,48 * new payload fields that are not purely additive, etc.).49 *50 * History:51 *   1 — initial release (session_start, session_end, full stream-json).52 */53export const DUAL_OUTPUT_PROTOCOL_VERSION = 1;54 55/**56 * Maximum bytes buffered in the Node.js WriteStream before the bridge57 * self-disables. Guards against unbounded memory growth when the output58 * target is a FIFO opened with O_RDWR (no EPIPE on reader disconnect).59 */60const MAX_BUFFERED_BYTES = 1024 * 1024; // 1 MB61 62/**63 * Optional metadata wired into the `session_start` capability handshake.64 */65export interface DualOutputBridgeOptions {66  /** CLI version string (e.g. "0.14.5"). Surfaced in session_start. */67  version?: string;68}69 70/**71 * Bridges TUI-mode events to a sidecar StreamJsonOutputAdapter that writes72 * structured JSON events to a secondary output channel (fd or file).73 *74 * This enables "dual output" mode: the TUI renders normally on stdout while75 * a parallel JSON event stream is emitted on a separate channel for76 * programmatic consumption by IDE extensions, web frontends, CI pipelines, etc.77 *78 * Usage:79 *   qwen --json-fd 3        # JSON events written to fd 380 *   qwen --json-file /path  # JSON events written to file/FIFO81 */82export class DualOutputBridge {83  private readonly adapter: StreamJsonOutputAdapter;84  private readonly stream: WriteStream;85  private readonly sessionId: string;86  private active = true;87  private shutdownPromise: Promise<void> | null = null;88 89  constructor(90    config: Config,91    target: { fd: number } | { filePath: string },92    options: DualOutputBridgeOptions = {},93  ) {94    this.sessionId = config.getSessionId();95    if ('fd' in target) {96      // Reject stdin/stdout/stderr to prevent corrupting TUI output97      if (target.fd <= 2) {98        throw new Error(99          `--json-fd ${target.fd}: file descriptors 0 (stdin), 1 (stdout), and 2 (stderr) ` +100            'are reserved. Use fd 3 or higher.',101        );102      }103      // Validate fd is open before attempting to use it104      try {105        fstatSync(target.fd);106      } catch {107        throw new Error(108          `--json-fd ${target.fd}: file descriptor is not open. ` +109            'The caller must provide this fd via spawn stdio configuration ' +110            'or shell redirection (e.g., 3>/tmp/events.jsonl).',111        );112      }113      this.stream = createWriteStream('', { fd: target.fd });114    } else {115      // Open with O_WRONLY|O_NONBLOCK to avoid blocking the event loop on FIFOs.116      // On FIFO, a regular open(O_WRONLY) blocks until a reader connects.117      // O_NONBLOCK makes openSync return immediately; if no reader is118      // connected yet (ENXIO), the catch block below retries with O_RDWR.119      try {120        const fd = openSync(121          target.filePath,122          constants.O_WRONLY | constants.O_NONBLOCK,123        );124        this.stream = createWriteStream('', { fd });125      } catch (err) {126        const code = (err as NodeJS.ErrnoException).code;127        if (code === 'ENXIO') {128          // FIFO with no reader connected yet. Use O_RDWR | O_NONBLOCK so129          // the open returns immediately (POSIX: process is both reader and130          // writer, satisfying the "at least one reader" requirement).131          // Trade-off: EPIPE won't fire on reader disconnect; the bridge132          // self-disables when the pipe buffer fills instead.133          try {134            const fd = openSync(135              target.filePath,136              constants.O_RDWR | constants.O_NONBLOCK,137            );138            this.stream = createWriteStream('', { fd });139          } catch (retryErr) {140            if ((retryErr as NodeJS.ErrnoException).code === 'EACCES') {141              throw new Error(142                `--json-file "${target.filePath}": permission denied opening FIFO for read-write. ` +143                  'Check read/write permissions on the file and its parent directories, ' +144                  'or start a reader before launching Qwen Code.',145              );146            }147            throw retryErr;148          }149        } else if (code === 'ENOENT') {150          // Regular file doesn't exist yet — create it.151          this.stream = createWriteStream(target.filePath, { flags: 'w' });152        } else {153          throw err;154        }155      }156    }157 158    this.stream.on('error', (err) => {159      const code = (err as NodeJS.ErrnoException).code;160      if (code === 'EPIPE' || code === 'ERR_STREAM_DESTROYED') {161        debugLogger.warn('DualOutput: consumer disconnected, disabling');162      } else if (code === 'ERR_SYSTEM_ERROR') {163        debugLogger.warn(164          'DualOutput: system error on stream, disabling:',165          (err as NodeJS.ErrnoException).message,166        );167      } else {168        debugLogger.error('DualOutput stream error:', err);169      }170      // Disable on any stream error to prevent repeated write failures171      this.active = false;172    });173 174    this.adapter = new StreamJsonOutputAdapter(175      config,176      true, // includePartialMessages — always emit streaming events177      this.stream,178    );179 180    // Announce the session immediately so consumers can correlate the channel181    // with a session before any other event arrives. The data payload also182    // serves as a capability handshake: consumers can read `protocol_version`183    // and `supported_events` to feature-detect without sniffing the stream.184    try {185      this.adapter.emitSystemMessage('session_start', {186        session_id: this.sessionId,187        cwd: process.cwd(),188        protocol_version: DUAL_OUTPUT_PROTOCOL_VERSION,189        version: options.version,190        supported_events: [...SUPPORTED_EVENTS],191      });192    } catch (err) {193      debugLogger.error('DualOutput session_start error:', err);194      this.active = false;195    }196  }197 198  processEvent(event: ServerGeminiStreamEvent): void {199    if (!this.active) return;200    this.disableIfBufferOverflowed();201    if (!this.active) return;202    try {203      this.adapter.processEvent(event);204    } catch (err) {205      debugLogger.error('DualOutput processEvent error:', err);206      this.active = false;207    }208  }209 210  startAssistantMessage(): void {211    if (!this.active) return;212    this.disableIfBufferOverflowed();213    if (!this.active) return;214    try {215      this.adapter.startAssistantMessage();216    } catch (err) {217      debugLogger.error('DualOutput startAssistantMessage error:', err);218      this.active = false;219    }220  }221 222  finalizeAssistantMessage(): void {223    if (!this.active) return;224    this.disableIfBufferOverflowed();225    if (!this.active) return;226    try {227      this.adapter.finalizeAssistantMessage();228    } catch (err) {229      debugLogger.error('DualOutput finalizeAssistantMessage error:', err);230      this.active = false;231    }232  }233 234  emitUserMessage(parts: Part[]): void {235    if (!this.active) return;236    this.disableIfBufferOverflowed();237    if (!this.active) return;238    try {239      this.adapter.emitUserMessage(parts);240    } catch (err) {241      debugLogger.error('DualOutput emitUserMessage error:', err);242      this.active = false;243    }244  }245 246  emitToolResult(247    request: ToolCallRequestInfo,248    response: ToolCallResponseInfo,249  ): void {250    if (!this.active) return;251    this.disableIfBufferOverflowed();252    if (!this.active) return;253    try {254      this.adapter.emitToolResult(request, response);255    } catch (err) {256      debugLogger.error('DualOutput emitToolResult error:', err);257      this.active = false;258    }259  }260 261  /** Whether the underlying stream is still writable. */262  get isConnected(): boolean {263    return this.active;264  }265 266  private disableIfBufferOverflowed(): void {267    if (this.stream.writableLength > MAX_BUFFERED_BYTES) {268      debugLogger.warn(269        'DualOutput: buffered data exceeds limit, disabling (no consumer draining?)',270      );271      this.active = false;272      this.stream.destroy();273    }274  }275 276  /**277   * Emits a `can_use_tool` permission request so an external consumer can278   * approve or deny the tool call. Pairs with {@link emitControlResponse}.279   */280  emitPermissionRequest(281    requestId: string,282    toolName: string,283    toolUseId: string,284    input: unknown,285    blockedPath: string | null = null,286  ): void {287    if (!this.active) return;288    this.disableIfBufferOverflowed();289    if (!this.active) return;290    try {291      this.adapter.emitPermissionRequest(292        requestId,293        toolName,294        toolUseId,295        input,296        blockedPath,297      );298    } catch (err) {299      debugLogger.error('DualOutput emitPermissionRequest error:', err);300      this.active = false;301    }302  }303 304  /**305   * Emits the result of a permission decision (made either in the TUI or by306   * the external consumer) so all observers stay in sync.307   */308  emitControlResponse(requestId: string, allowed: boolean): void {309    if (!this.active) return;310    this.disableIfBufferOverflowed();311    if (!this.active) return;312    try {313      this.adapter.emitControlResponse(requestId, allowed);314    } catch (err) {315      debugLogger.error('DualOutput emitControlResponse error:', err);316      this.active = false;317    }318  }319 320  /**321   * Emits a `control_response` with subtype `error` — used when an external322   * `confirmation_response` cannot be satisfied (unknown request_id, the323   * tool call already resolved, stream already closed, etc.). Lets324   * consumers retry or surface the error instead of silently hanging.325   */326  emitControlError(requestId: string, message: string): void {327    if (!this.active) return;328    this.disableIfBufferOverflowed();329    if (!this.active) return;330    try {331      this.adapter.emitControlError(requestId, message);332    } catch (err) {333      debugLogger.error('DualOutput emitControlError error:', err);334      this.active = false;335    }336  }337 338  /** General-purpose system event escape hatch. */339  emitSystemMessage(subtype: string, data?: unknown): void {340    if (!this.active) return;341    this.disableIfBufferOverflowed();342    if (!this.active) return;343    try {344      this.adapter.emitSystemMessage(subtype, data);345    } catch (err) {346      debugLogger.error('DualOutput emitSystemMessage error:', err);347      this.active = false;348    }349  }350 351  shutdown(): Promise<void> {352    if (this.shutdownPromise) return this.shutdownPromise;353    // Try to emit session_end before tearing the stream down so consumers354    // get a definitive termination signal rather than inferring it from355    // EPIPE. Failures here are swallowed — the stream may already be in an356    // error state if the consumer disconnected first.357    if (this.active) {358      try {359        this.adapter.emitSystemMessage('session_end', {360          session_id: this.sessionId,361        });362      } catch {363        // ignore — stream likely already closed364      }365    }366    this.active = false;367    this.shutdownPromise = new Promise((resolve) => {368      if (this.stream.closed || this.stream.destroyed) {369        resolve();370        return;371      }372 373      const cleanup = () => {374        this.stream.off('close', onClose);375        this.stream.off('error', onError);376      };377      const onClose = () => {378        cleanup();379        resolve();380      };381      const onError = (err: Error) => {382        debugLogger.debug('DualOutput: stream error during shutdown:', err);383      };384 385      this.stream.once('close', onClose);386      this.stream.once('error', onError);387 388      try {389        this.stream.end();390      } catch (err) {391        cleanup();392        debugLogger.debug('DualOutput: stream end error during shutdown:', err);393        resolve();394      }395    });396    return this.shutdownPromise;397  }398}399