pbakaus_impeccable/tests/live-poll-stream.test.mjs

/**
 * Integration tests for live-poll --stream mode.
 * Run with: node --test tests/live-poll-stream.test.mjs
 */

import { describe, it, before, after } from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, readFileSync, rmSync } from 'node:fs';
import { join } from 'node:path';
import { spawn } from 'node:child_process';
import { tmpdir } from 'node:os';
import { getLiveServerPath } from '../skill/scripts/lib/impeccable-paths.mjs';
import { postReply } from '../skill/scripts/live-poll.mjs';

const REPO_ROOT = process.cwd();
const SERVER_SCRIPT = join(REPO_ROOT, 'skill/scripts/live-server.mjs');
const POLL_SCRIPT = join(REPO_ROOT, 'skill/scripts/live-poll.mjs');

function startServer(port = 8498, { cwd = REPO_ROOT } = {}) {
  return new Promise((resolve, reject) => {
    const proc = spawn('node', [SERVER_SCRIPT, '--port=' + port], {
      cwd,
      stdio: ['ignore', 'pipe', 'pipe'],
      env: { ...process.env },
    });
    let output = '';
    proc.stdout.on('data', (d) => {
      output += d.toString();
      if (output.includes('running on')) {
        try {
          const info = JSON.parse(readFileSync(getLiveServerPath(cwd), 'utf-8'));
          resolve({ proc, port: info.port, token: info.token, cwd });
        } catch {
          reject(new Error('Server started but PID file not readable'));
        }
      }
    });
    proc.stderr.on('data', (d) => { output += d.toString(); });
    proc.on('error', reject);
    setTimeout(() => reject(new Error('Server start timeout. Output: ' + output)), 5000);
  });
}

async function stopServer(port, token) {
  try {
    await fetch(`http://localhost:${port}/stop?token=${token}`);
  } catch { /* already gone */ }
}

function readStdoutLine(streamProc, timeoutMs = 8000) {
  return new Promise((resolve, reject) => {
    let buffer = '';
    const timer = setTimeout(() => {
      cleanup();
      reject(new Error('Timed out waiting for stdout line. Buffer: ' + buffer.slice(0, 200)));
    }, timeoutMs);

    function onData(chunk) {
      buffer += chunk.toString();
      const idx = buffer.indexOf('\n');
      if (idx === -1) return;
      const line = buffer.slice(0, idx).trim();
      buffer = buffer.slice(idx + 1);
      cleanup();
      resolve(line);
    }

    function cleanup() {
      clearTimeout(timer);
      streamProc.stdout.off('data', onData);
    }

    streamProc.stdout.on('data', onData);
    if (buffer.includes('\n')) onData('');
  });
}

describe('live-poll --stream integration', () => {
  let server;
  let serverCwd;

  before(async () => {
    serverCwd = mkdtempSync(join(tmpdir(), 'impeccable-live-poll-stream-'));
    server = await startServer(8498, { cwd: serverCwd });
  });

  after(async () => {
    if (server?.proc && !server.proc.killed) {
      await stopServer(server.port, server.token);
      server.proc.kill('SIGTERM');
    }
    if (serverCwd) rmSync(serverCwd, { recursive: true, force: true });
  });

  it('emits multiple steer events without restarting the poll process', async () => {
    const streamProc = spawn('node', [POLL_SCRIPT, '--stream', '--ack-timeout=15000'], {
      cwd: server.cwd,
      stdio: ['ignore', 'pipe', 'pipe'],
      env: { ...process.env },
    });

    try {
      const firstLinePromise = readStdoutLine(streamProc);

      await new Promise((r) => setTimeout(r, 150));

      const post1 = await fetch(`http://localhost:${server.port}/events`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          token: server.token,
          type: 'steer',
          id: '11111111',
          message: 'stream test one',
          pageUrl: 'http://localhost:4321/',
        }),
      });
      assert.equal(post1.status, 200);

      const firstLine = await firstLinePromise;
      const firstEvent = JSON.parse(firstLine);
      assert.equal(firstEvent.type, 'steer');
      assert.equal(firstEvent.id, '11111111');

      await postReply(`http://localhost:${server.port}`, server.token, {
        id: '11111111',
        type: 'steer_done',
        message: 'done one',
      });

      const secondLinePromise = readStdoutLine(streamProc);

      const post2 = await fetch(`http://localhost:${server.port}/events`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          token: server.token,
          type: 'steer',
          id: '22222222',
          message: 'stream test two',
          pageUrl: 'http://localhost:4321/',
        }),
      });
      assert.equal(post2.status, 200);

      const secondLine = await secondLinePromise;
      const secondEvent = JSON.parse(secondLine);
      assert.equal(secondEvent.type, 'steer');
      assert.equal(secondEvent.id, '22222222');
      assert.equal(secondEvent.message, 'stream test two');
    } finally {
      streamProc.kill('SIGTERM');
    }
  });

  it('emits insert-mode generate and clears pending after done reply', async () => {
    const streamProc = spawn('node', [POLL_SCRIPT, '--stream', '--ack-timeout=15000'], {
      cwd: server.cwd,
      stdio: ['ignore', 'pipe', 'pipe'],
      env: { ...process.env },
    });

    try {
      const linePromise = readStdoutLine(streamProc);

      await new Promise((r) => setTimeout(r, 150));

      const postRes = await fetch(`http://localhost:${server.port}/events`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          token: server.token,
          type: 'generate',
          id: 'aa999999',
          mode: 'insert',
          count: 3,
          pageUrl: 'http://localhost:4321/',
          insert: {
            position: 'after',
            anchor: { tagName: 'section', classes: ['hero'] },
          },
          placeholder: { width: 320, height: 80 },
          freeformPrompt: 'Add testimonials',
        }),
      });
      assert.equal(postRes.status, 200);

      const line = await linePromise;
      const event = JSON.parse(line);
      assert.equal(event.type, 'generate');
      assert.equal(event.mode, 'insert');
      assert.equal(event.id, 'aa999999');
      assert.equal(event.freeformPrompt, 'Add testimonials');

      await postReply(`http://localhost:${server.port}`, server.token, {
        id: 'aa999999',
        type: 'done',
      });
    } finally {
      streamProc.kill('SIGTERM');
    }
  });
});