feat: add streaming API interceptor

Author: AstaTus

src/browser/base-page.ts

@@ -18,6 +18,7 @@ import {
   waitForTextJs,
   waitForCaptureJs,
   waitForSelectorJs,
+  waitForStreamCaptureJs,
   scrollJs,
   autoScrollJs,
   networkRequestsJs,
@@ -159,6 +160,22 @@ export abstract class BasePage implements IPage {
     await this.evaluate(waitForCaptureJs(maxMs));
   }
 
+  async installStreamingInterceptor(pattern: string): Promise<void> {
+    const { generateStreamingInterceptorJs } = await import('../interceptor.js');
+    await this.evaluate(generateStreamingInterceptorJs(JSON.stringify(pattern)));
+  }
+
+  async getStreamedResponses(opts?: { clear?: boolean }): Promise<{ text: string; events: any[]; done: boolean; errors: any[] }> {
+    const { generateReadStreamJs } = await import('../interceptor.js');
+    const clear = opts?.clear !== false; // default true for backwards compat
+    return (await this.evaluate(generateReadStreamJs('__opencli_stream', clear))) as { text: string; events: any[]; done: boolean; errors: any[] };
+  }
+
+  async waitForStreamCapture(timeout: number = 30, opts?: { minChars?: number; waitForDone?: boolean }): Promise<void> {
+    const maxMs = timeout * 1000;
+    await this.evaluate(waitForStreamCaptureJs(maxMs, opts));
+  }
+
   /** Fallback basic snapshot */
   protected async _basicSnapshot(opts: Pick<SnapshotOptions, 'interactive' | 'compact' | 'maxDepth' | 'raw'> = {}): Promise<unknown> {
     const maxDepth = Math.max(1, Math.min(Number(opts.maxDepth) || 50, 200));

src/browser/cdp.ts

@@ -15,7 +15,7 @@ import type { BrowserCookie, IPage, ScreenshotOptions } from '../types.js';
 import type { IBrowserFactory } from '../runtime.js';
 import { wrapForEval } from './utils.js';
 import { generateStealthJs } from './stealth.js';
-import { waitForDomStableJs } from './dom-helpers.js';
+import { waitForDomStableJs, waitForCaptureJs } from './dom-helpers.js';
 import { isRecord, saveBase64ToFile } from '../utils.js';
 import { getAllElectronApps } from '../electron-apps.js';
 import { BasePage } from './base-page.js';
@@ -234,6 +234,33 @@ class CDPPage extends BasePage {
   async selectTab(_index: number): Promise<void> {
     // Not supported in direct CDP mode
   }
+
+  async consoleMessages(_level?: string): Promise<unknown[]> {
+    return [];
+  }
+
+  async getCurrentUrl(): Promise<string | null> {
+    return this._lastUrl;
+  }
+
+  async installInterceptor(pattern: string): Promise<void> {
+    const { generateInterceptorJs } = await import('../interceptor.js');
+    await this.evaluate(generateInterceptorJs(JSON.stringify(pattern), {
+      arrayName: '__opencli_xhr',
+      patchGuard: '__opencli_interceptor_patched',
+    }));
+  }
+
+  async getInterceptedRequests(): Promise<unknown[]> {
+    const { generateReadInterceptedJs } = await import('../interceptor.js');
+    const result = await this.evaluate(generateReadInterceptedJs('__opencli_xhr'));
+    return Array.isArray(result) ? result : [];
+  }
+
+  async waitForCapture(timeout: number = 10): Promise<void> {
+    const maxMs = timeout * 1000;
+    await this.evaluate(waitForCaptureJs(maxMs));
+  }
 }
 
 function isCookie(value: unknown): value is BrowserCookie {

src/browser/daemon-client.ts

@@ -90,6 +90,7 @@ export async function isExtensionConnected(): Promise<boolean> {
 export async function sendCommand(
   action: DaemonCommand['action'],
   params: Omit<DaemonCommand, 'id' | 'action'> = {},
+  timeoutMs: number = 30000,
 ): Promise<unknown> {
   const maxRetries = 4;
 
@@ -99,7 +100,7 @@ export async function sendCommand(
     const command: DaemonCommand = { id, action, ...params };
     try {
       const controller = new AbortController();
-      const timer = setTimeout(() => controller.abort(), 30000);
+      const timer = setTimeout(() => controller.abort(), timeoutMs);
 
       const res = await fetch(`${DAEMON_URL}/command`, {
         method: 'POST',

src/browser/dom-helpers.test.ts

@@ -1,5 +1,5 @@
 import { describe, it, expect } from 'vitest';
-import { autoScrollJs, waitForCaptureJs, waitForSelectorJs } from './dom-helpers.js';
+import { autoScrollJs, waitForCaptureJs, waitForSelectorJs, waitForStreamCaptureJs } from './dom-helpers.js';
 
 describe('autoScrollJs', () => {
   it('returns early without error when document.body is null', async () => {
@@ -112,3 +112,64 @@ describe('waitForSelectorJs', () => {
     delete g.MutationObserver;
   });
 });
+
+describe('waitForStreamCaptureJs', () => {
+  it('returns a non-empty string with default prefix', () => {
+    const code = waitForStreamCaptureJs(1000);
+    expect(typeof code).toBe('string');
+    expect(code.length).toBeGreaterThan(0);
+    expect(code).toContain('__opencli_stream_text');
+    expect(code).toContain('__opencli_stream_done');
+  });
+
+  it('generates code that resolves when minChars is reached', async () => {
+    const g = globalThis as any;
+    g.__opencli_stream_text = '';
+    g.__opencli_stream_done = false;
+    g.window = g;
+    const code = waitForStreamCaptureJs(1000, { minChars: 5 });
+    const promise = eval(code) as Promise<void>;
+    // Simulate data arriving
+    g.__opencli_stream_text = 'hello world';
+    await expect(promise).resolves.not.toThrow();
+    delete g.__opencli_stream_text;
+    delete g.__opencli_stream_done;
+    delete g.window;
+  });
+
+  it('generates code that resolves when done flag is set', async () => {
+    const g = globalThis as any;
+    g.__opencli_stream_text = '';
+    g.__opencli_stream_done = false;
+    g.window = g;
+    const code = waitForStreamCaptureJs(1000, { waitForDone: true });
+    const promise = eval(code) as Promise<void>;
+    // Simulate stream completion — need both minChars AND done
+    g.__opencli_stream_text = 'data arrived';
+    g.__opencli_stream_done = true;
+    await expect(promise).resolves.not.toThrow();
+    delete g.__opencli_stream_text;
+    delete g.__opencli_stream_done;
+    delete g.window;
+  });
+
+  it('generates code that rejects on timeout', async () => {
+    const g = globalThis as any;
+    g.__opencli_stream_text = '';
+    g.__opencli_stream_done = false;
+    g.window = g;
+    const code = waitForStreamCaptureJs(50, { minChars: 100, waitForDone: true });
+    const promise = eval(code) as Promise<void>;
+    await expect(promise).rejects.toThrow();
+    delete g.__opencli_stream_text;
+    delete g.__opencli_stream_done;
+    delete g.window;
+  });
+
+  it('uses custom prefix when provided', () => {
+    const code = waitForStreamCaptureJs(1000, { prefix: '__my_prefix' });
+    expect(code).toContain('__my_prefix_text');
+    expect(code).toContain('__my_prefix_done');
+    expect(code).not.toContain('__opencli_stream');
+  });
+});

src/browser/dom-helpers.ts

@@ -200,6 +200,35 @@ export function waitForCaptureJs(maxMs: number): string {
   `;
 }
 
+/**
+ * Generate JS to wait until the streaming interceptor has captured data.
+ * Polls window.__opencli_stream_text length. Optionally waits for stream completion.
+ * 50ms interval, rejects after maxMs.
+ */
+export function waitForStreamCaptureJs(
+  maxMs: number,
+  opts: { minChars?: number; waitForDone?: boolean; prefix?: string } = {},
+): string {
+  const minChars = opts.minChars ?? 1;
+  const waitForDone = opts.waitForDone ?? false;
+  const prefix = opts.prefix ?? '__opencli_stream';
+  return `
+    new Promise((resolve, reject) => {
+      const deadline = Date.now() + ${maxMs};
+      const check = () => {
+        const text = window.${prefix}_text || '';
+        const done = window.${prefix}_done || false;
+        if (text.length >= ${minChars} && (${waitForDone} ? done : true)) {
+          return resolve('captured');
+        }
+        if (Date.now() > deadline) return reject(new Error('Stream capture timeout'));
+        setTimeout(check, 50);
+      };
+      check();
+    })
+  `;
+}
+
 /**
  * Generate JS to wait until document.querySelector(selector) returns a match.
  * Uses MutationObserver for near-instant resolution; falls back to reject after timeoutMs.

src/browser/page.ts

@@ -15,9 +15,9 @@ import { sendCommand } from './daemon-client.js';
 import { wrapForEval } from './utils.js';
 import { saveBase64ToFile } from '../utils.js';
 import { generateStealthJs } from './stealth.js';
-import { waitForDomStableJs } from './dom-helpers.js';
-import { BasePage } from './base-page.js';
 
+import { waitForDomStableJs, waitForCaptureJs, waitForStreamCaptureJs } from './dom-helpers.js';
+import { BasePage } from './base-page.js';
 export function isRetryableSettleError(err: unknown): boolean {
   const message = err instanceof Error ? err.message : String(err);
   return message.includes('Inspected target navigated or closed')
@@ -179,6 +179,15 @@ export class Page extends BasePage {
       throw new Error('setFileInput returned no count — command may not be supported by the extension');
     }
   }
+
+  // Override: waitForStreamCapture needs longer HTTP timeout for long-running browser promises
+  async waitForStreamCapture(timeout: number = 30, opts?: { minChars?: number; waitForDone?: boolean }): Promise<void> {
+    const maxMs = timeout * 1000;
+    await sendCommand('exec', {
+      code: waitForStreamCaptureJs(maxMs, opts),
+      ...this._cmdOpts(),
+    }, maxMs + 10000); // HTTP timeout = browser timeout + 10s buffer
+  }
 }
 
 // (End of file)

src/interceptor.test.ts

@@ -3,7 +3,7 @@
  */
 
 import { describe, it, expect } from 'vitest';
-import { generateInterceptorJs, generateReadInterceptedJs, generateTapInterceptorJs } from './interceptor.js';
+import { generateInterceptorJs, generateReadInterceptedJs, generateTapInterceptorJs, generateStreamingInterceptorJs, generateReadStreamJs } from './interceptor.js';
 
 describe('generateInterceptorJs', () => {
   it('generates valid JavaScript function source', () => {
@@ -92,3 +92,132 @@ describe('generateTapInterceptorJs', () => {
     expect(tap.xhrPatch).toContain('!captured');
   });
 });
+
+describe('generateStreamingInterceptorJs', () => {
+  it('generates valid JavaScript function source', () => {
+    const js = generateStreamingInterceptorJs('"api/stream"');
+    expect(js.trim()).toMatch(/^\(\)\s*=>/);
+    expect(js).toContain('"api/stream"');
+  });
+
+  it('initializes all streaming state variables', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    expect(js).toContain('__opencli_stream_text');
+    expect(js).toContain('__opencli_stream_events');
+    expect(js).toContain('__opencli_stream_done');
+    expect(js).toContain('__opencli_stream_errors');
+    expect(js).toContain('__opencli_stream_sse_buf');
+  });
+
+  it('resets SSE buffer on install', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    // SSE buffer should be reset at initialization time
+    expect(js).toContain('__opencli_stream_sse_buf');
+    expect(js).toContain("= ''");
+  });
+
+  it('resets all state including SSE buffer on each new matching fetch', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    // Inside the fetch patch, SSE buffer should be cleared for new request
+    const fetchResetCount = (js.match(/__opencli_stream_sse_buf/g) || []).length;
+    // Should appear at least twice: init + fetch reset + XHR reset
+    expect(fetchResetCount).toBeGreaterThanOrEqual(2);
+  });
+
+  it('patches both fetch and XHR', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    expect(js).toContain('window.fetch');
+    expect(js).toContain('XMLHttpRequest.prototype');
+    expect(js).toContain('onprogress');
+    expect(js).toContain('readystatechange');
+  });
+
+  it('normalizes CRLF in SSE parsing', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    // SSE parser should normalize \r\n to \n
+    expect(js).toContain(String.raw`replace(/\r\n/g, '\n')`);
+  });
+
+  it('normalizes CRLF in fetch SSE boundary detection', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    // Fetch path should normalize CRLF before SSE boundary splitting
+    expect(js).toContain('sseBuffer');
+    expect(js).toContain(String.raw`replace(/\r\n/g, '\n')`);
+  });
+
+  it('normalizes CRLF in XHR SSE boundary detection', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    // XHR progress path should normalize CRLF
+    expect(js).toContain(String.raw`chunk.replace(/\r\n/g, '\n')`);
+  });
+
+  it('includes SSE parser function', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    expect(js).toContain('__parseSse');
+    expect(js).toContain('event:');
+    expect(js).toContain('data:');
+  });
+
+  it('uses custom prefix and patch guard', () => {
+    const js = generateStreamingInterceptorJs('"test"', {
+      arrayName: '__my_stream',
+      patchGuard: '__my_stream_guard',
+    });
+    expect(js).toContain('__my_stream_text');
+    expect(js).toContain('__my_stream_events');
+    expect(js).toContain('__my_stream_guard');
+    expect(js).not.toContain('__opencli_stream');
+  });
+
+  it('uses custom maxChunks', () => {
+    const js = generateStreamingInterceptorJs('"test"', { maxChunks: 100 });
+    expect(js).toContain('100');
+  });
+
+  it('XHR readystatechange uses authoritative responseText overwrite', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    // readystatechange readyState=4 should overwrite with full responseText
+    expect(js).toContain('readyState === 4');
+    expect(js).toContain('window.__opencli_stream_text = full');
+  });
+
+  it('XHR load event is guarded by settled flag', () => {
+    const js = generateStreamingInterceptorJs('"test"');
+    expect(js).toContain('if (settled) return');
+  });
+});
+
+describe('generateReadStreamJs', () => {
+  it('generates valid JavaScript to read streaming state', () => {
+    const js = generateReadStreamJs();
+    expect(js.trim()).toMatch(/^\(\)\s*=>/);
+    expect(js).toContain('__opencli_stream_text');
+    expect(js).toContain('__opencli_stream_events');
+    expect(js).toContain('__opencli_stream_done');
+    expect(js).toContain('__opencli_stream_errors');
+  });
+
+  it('clears all state including SSE buffer by default', () => {
+    const js = generateReadStreamJs();
+    expect(js).toContain('__opencli_stream_text');
+    expect(js).toContain('__opencli_stream_sse_buf');
+    expect(js).toContain("= ''");
+    expect(js).toContain('= []');
+  });
+
+  it('does not clear state when clear=false (peek mode)', () => {
+    const js = generateReadStreamJs('__opencli_stream', false);
+    // Should NOT contain clearing statements
+    expect(js).not.toContain("window.__opencli_stream_text = ''");
+    expect(js).not.toContain("window.__opencli_stream_sse_buf = ''");
+    // But should still contain the read statements
+    expect(js).toContain('__opencli_stream_text');
+  });
+
+  it('uses custom prefix', () => {
+    const js = generateReadStreamJs('__my_prefix');
+    expect(js).toContain('__my_prefix_text');
+    expect(js).toContain('__my_prefix_events');
+    expect(js).not.toContain('__opencli_stream');
+  });
+});