basant307/AI_Governance_Project

1/**2 * @license3 * Copyright 2025 Qwen Team4 * SPDX-License-Identifier: Apache-2.05 */6 7import { createHash } from 'node:crypto';8import {9  closeSync,10  createReadStream,11  openSync,12  readSync,13  statSync,14  unwatchFile,15  watchFile,16} from 'node:fs';17import { createInterface } from 'node:readline';18import { createDebugLogger } from '@qwen-code/qwen-code-core';19 20const debugLogger = createDebugLogger('REMOTE_INPUT');21 22/**23 * JSONL command shapes written by an external process (IDE extension,24 * web frontend, automation script) into the file passed to --input-file.25 *26 * - `submit`: enqueue a user message that the TUI processes as if typed27 *   into the prompt.28 * - `confirmation_response`: reply to a pending tool-permission29 *   `control_request` previously emitted on the dual-output channel.30 */31export type RemoteInputCommand =32  | { type: 'submit'; text: string }33  | { type: 'confirmation_response'; request_id: string; allowed: boolean };34 35/**36 * Callback invoked when a `confirmation_response` command is read.37 */38export type ConfirmationHandler = (requestId: string, allowed: boolean) => void;39 40/**41 * Callback type for submitting a query from remote input.42 * Returns true if the submit was accepted, false if rejected (TUI busy).43 */44export type SubmitFn = (45  query: string,46) => Promise<boolean | void> | boolean | void;47 48/**49 * Watches a JSONL file for remote input commands and calls the registered50 * submit function when new commands arrive.51 *52 * The watcher queues commands and retries when the TUI is busy (responding).53 * Call `notifyIdle()` when the TUI transitions to idle state to trigger54 * processing of queued commands.55 */56export class RemoteInputWatcher {57  private submitFn: SubmitFn | null = null;58  private confirmationHandler: ConfirmationHandler | null = null;59  private queue: Array<Extract<RemoteInputCommand, { type: 'submit' }>> = [];60  private processing = false;61  private active = true;62  private bytesRead = 0;63  private consumedPrefixHash: string | null = null;64  private reading = false;65  private filePath: string;66  private retryTimer: ReturnType<typeof setTimeout> | null = null;67  private readonly pollIntervalMs: number;68 69  constructor(filePath: string, options?: { pollIntervalMs?: number }) {70    this.filePath = filePath;71    this.pollIntervalMs = options?.pollIntervalMs ?? 500;72    this.startWatching();73  }74 75  /**76   * Register the TUI's submit function. Called from AppContainer77   * once useGeminiStream's submitQuery is available.78   */79  setSubmitFn(fn: SubmitFn): void {80    this.submitFn = fn;81    this.processQueue();82  }83 84  /**85   * Register the handler invoked when a `confirmation_response` command is86   * read from the input file. Used to bridge external approvals back into87   * the tool's `onConfirm` callback.88   */89  setConfirmationHandler(fn: ConfirmationHandler): void {90    this.confirmationHandler = fn;91  }92 93  /**94   * Notify the watcher that the TUI has become idle.95   * Call this when streamingState transitions to Idle — it triggers96   * processing of any queued commands that were deferred due to TUI busy.97   */98  notifyIdle(): void {99    if (this.queue.length > 0 && !this.processing) {100      this.processQueue();101    }102  }103 104  private startWatching(): void {105    try {106      const stat = statSync(this.filePath);107      this.bytesRead = stat.size;108      this.consumedPrefixHash = this.hashFilePrefix(this.bytesRead);109    } catch {110      this.bytesRead = 0;111      this.consumedPrefixHash = null;112    }113 114    watchFile(this.filePath, { interval: this.pollIntervalMs }, () => {115      if (!this.active) return;116      this.readNewLines();117    });118 119    debugLogger.debug(`RemoteInput: watching ${this.filePath}`);120  }121 122  /**123   * Manually trigger a check for new input. Returns a promise that resolves124   * once any new lines have been read and processed. In production the125   * `watchFile` poll calls this automatically; tests can call it directly126   * to avoid depending on filesystem-polling timing.127   */128  checkForNewInput(): Promise<void> {129    return this.readNewLines();130  }131 132  private readNewLines(): Promise<void> {133    if (!this.active || this.reading) return Promise.resolve();134 135    let currentSize: number;136    try {137      const stat = statSync(this.filePath);138      currentSize = stat.size;139    } catch {140      return Promise.resolve();141    }142 143    // Size alone misses truncate+rewrite that lands at the same or a larger144    // size. Append-only writes preserve the consumed prefix hash; rewrites do not.145    if (currentSize < this.bytesRead) {146      debugLogger.debug(147        'RemoteInput: input file shrank, resetting read offset',148      );149      this.bytesRead = 0;150      this.consumedPrefixHash = null;151    } else if (this.hasConsumedPrefixChanged()) {152      debugLogger.debug(153        'RemoteInput: input file prefix changed, resetting read offset',154      );155      this.bytesRead = 0;156      this.consumedPrefixHash = null;157    }158 159    if (currentSize <= this.bytesRead) return Promise.resolve();160 161    const consumeUntil = this.findLastCompleteRecordEnd(162      this.bytesRead,163      currentSize,164    );165    if (consumeUntil === null) {166      return Promise.resolve();167    }168 169    const nextConsumedPrefixHash = this.hashFilePrefix(consumeUntil);170    this.reading = true;171    const stream = createReadStream(this.filePath, {172      start: this.bytesRead,173      end: consumeUntil - 1,174      encoding: 'utf-8',175    });176    const rl = createInterface({ input: stream, crlfDelay: Infinity });177 178    rl.on('line', (line) => {179      const trimmed = line.trim();180      if (!trimmed) return;181      try {182        const cmd = JSON.parse(trimmed);183        // confirmation_response is dispatched immediately rather than queued:184        // a pending tool call is blocking and the response must reach185        // onConfirm without waiting for any earlier `submit` to finish.186        if (187          cmd &&188          cmd.type === 'confirmation_response' &&189          typeof cmd.request_id === 'string' &&190          typeof cmd.allowed === 'boolean'191        ) {192          debugLogger.debug(193            `RemoteInput: confirmation_response for ${cmd.request_id} (allowed=${cmd.allowed})`,194          );195          this.confirmationHandler?.(cmd.request_id, cmd.allowed);196        } else if (197          cmd &&198          cmd.type === 'submit' &&199          typeof cmd.text === 'string'200        ) {201          debugLogger.debug(202            `RemoteInput: queued command: ${cmd.text.slice(0, 50)}...`,203          );204          this.queue.push(205            cmd as Extract<RemoteInputCommand, { type: 'submit' }>,206          );207        } else {208          debugLogger.warn(209            `RemoteInput: unknown command type: ${String(cmd?.type)}`,210          );211        }212      } catch (_err) {213        debugLogger.warn(`RemoteInput: failed to parse line: ${trimmed}`);214      }215    });216 217    return new Promise<void>((resolve) => {218      rl.on('close', () => {219        this.bytesRead = consumeUntil;220        if (nextConsumedPrefixHash !== null) {221          this.consumedPrefixHash = nextConsumedPrefixHash;222        }223        this.reading = false;224        this.processQueue();225        resolve();226      });227    });228  }229 230  private findLastCompleteRecordEnd(start: number, end: number): number | null {231    if (end <= start) return null;232 233    let fd: number | null = null;234    try {235      fd = openSync(this.filePath, 'r');236      const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, end - start));237      let remaining = end - start;238      let position = start;239      let lastLineBreak = -1;240 241      while (remaining > 0) {242        const bytesToRead = Math.min(buffer.length, remaining);243        const bytesRead = readSync(fd, buffer, 0, bytesToRead, position);244        if (bytesRead <= 0) break;245        for (let i = 0; i < bytesRead; i++) {246          if (buffer[i] === 0x0a) {247            lastLineBreak = position + i;248          }249        }250        remaining -= bytesRead;251        position += bytesRead;252      }253 254      return lastLineBreak >= start ? lastLineBreak + 1 : null;255    } catch (err) {256      debugLogger.warn('RemoteInput: failed to scan complete records:', err);257      return null;258    } finally {259      if (fd !== null) {260        closeSync(fd);261      }262    }263  }264 265  private hasConsumedPrefixChanged(): boolean {266    if (this.bytesRead === 0) {267      return false;268    }269    if (this.consumedPrefixHash === null) {270      debugLogger.warn(271        'RemoteInput: missing consumed prefix hash, resetting read offset',272      );273      return true;274    }275 276    const currentHash = this.hashFilePrefix(this.bytesRead);277    if (currentHash === null) {278      debugLogger.warn(279        'RemoteInput: failed to hash consumed prefix, resetting read offset',280      );281      return true;282    }283    return currentHash !== this.consumedPrefixHash;284  }285 286  private hashFilePrefix(size: number): string | null {287    if (size <= 0) return null;288 289    let fd: number | null = null;290    try {291      fd = openSync(this.filePath, 'r');292      const hash = createHash('sha256');293      const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, size));294      let remaining = size;295      let position = 0;296 297      while (remaining > 0) {298        const bytesToRead = Math.min(buffer.length, remaining);299        const bytesRead = readSync(fd, buffer, 0, bytesToRead, position);300        if (bytesRead <= 0) return null;301        hash.update(buffer.subarray(0, bytesRead));302        remaining -= bytesRead;303        position += bytesRead;304      }305 306      return hash.digest('base64');307    } catch (err) {308      debugLogger.warn('RemoteInput: failed to hash file prefix:', err);309      return null;310    } finally {311      if (fd !== null) {312        closeSync(fd);313      }314    }315  }316 317  private async processQueue(): Promise<void> {318    if (this.processing || !this.submitFn || this.queue.length === 0) return;319 320    this.processing = true;321    if (this.retryTimer) {322      clearTimeout(this.retryTimer);323      this.retryTimer = null;324    }325 326    try {327      while (this.queue.length > 0 && this.active) {328        if (!this.submitFn) break;329        const cmd = this.queue[0]!; // peek, don't shift yet330        debugLogger.debug(331          `RemoteInput: submitting: ${cmd.text.slice(0, 50)}...`,332        );333        try {334          const result = await this.submitFn(cmd.text);335          // If submitFn returns false explicitly, the TUI rejected it (busy)336          if (result === false) {337            debugLogger.debug('RemoteInput: TUI busy, will retry on idle');338            this.scheduleRetry();339            break;340          }341          // Success — remove from queue342          this.queue.shift();343        } catch (err) {344          debugLogger.error('RemoteInput: submit failed:', err);345          this.queue.shift(); // remove failed command to avoid infinite retry346        }347        // Small delay between commands to let the TUI process348        if (this.queue.length > 0) {349          await new Promise((r) => setTimeout(r, 500));350        }351      }352    } finally {353      this.processing = false;354    }355  }356 357  private scheduleRetry(): void {358    if (this.retryTimer) return;359    // Retry after 2s if notifyIdle hasn't been called360    this.retryTimer = setTimeout(() => {361      this.retryTimer = null;362      if (this.queue.length > 0 && !this.processing) {363        this.processQueue();364      }365    }, 2000);366  }367 368  shutdown(): void {369    this.active = false;370    unwatchFile(this.filePath);371    if (this.retryTimer) clearTimeout(this.retryTimer);372    this.queue.length = 0;373    debugLogger.debug('RemoteInput: shut down');374  }375}376