pbakaus_impeccable/tests/live-codex-worker-supervisor.test.mjs

import assert from 'node:assert/strict';
import { mkdtempSync, mkdirSync, readFileSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { describe, it } from 'node:test';

import { CODEX_WORKER_OWNER } from '../skill/scripts/live/codex-worker.mjs';
import {
  CODEX_WORKER_EVENT_LEASE_MS,
  CODEX_WORKER_EVENT_TYPES,
  CodexLiveWorkerSupervisor,
  buildDetectorRepairPrompt,
  buildDeterministicScaffoldCommand,
  resolveDetectorFindingWaivers,
} from '../skill/scripts/live/codex-worker-supervisor.mjs';
import { createLiveSessionStore } from '../skill/scripts/live/session-store.mjs';
import { selectAvailablePendingEvent } from '../skill/scripts/live/poll-lanes.mjs';

describe('Codex Live worker supervisor ownership and lifecycle', () => {
  it('partitions worker and foreground control events without overlapping leases', () => {
    const entries = [
      { event: { type: 'steer' }, leaseUntil: 0, seq: 1 },
      { event: { type: 'generate' }, leaseUntil: 0, seq: 2 },
      { event: { type: 'manual_edit_apply' }, leaseUntil: 0, seq: 3 },
      { event: { type: 'accept' }, leaseUntil: 0, seq: 4 },
      { event: { type: 'carbonize_cleanup' }, leaseUntil: 0, seq: 5 },
      { event: { type: 'exit' }, leaseUntil: 0, seq: 6 },
    ];
    assert.equal(selectAvailablePendingEvent(entries, { types: CODEX_WORKER_EVENT_TYPES }).event.type, 'accept');
    assert.equal(selectAvailablePendingEvent(entries, {
      types: ['steer', 'manual_edit_apply', 'carbonize_cleanup', 'exit'],
    }).event.type, 'exit');
    assert.equal(CODEX_WORKER_EVENT_TYPES.includes('steer'), false);
    assert.equal(CODEX_WORKER_EVENT_TYPES.includes('manual_edit_apply'), false);
    assert.equal(CODEX_WORKER_EVENT_TYPES.includes('carbonize_cleanup'), false);
    assert.equal(CODEX_WORKER_EVENT_TYPES.includes('exit'), false);
  });

  it('builds the same deterministic wrap/insert target contract as foreground Live', () => {
    const replace = buildDeterministicScaffoldCommand({
      id: 'abc12345',
      count: 3,
      element: { id: 'hero', classes: ['hero', 'title'], tagName: 'H1', textContent: '  Exact hero copy  ' },
    }, '/scripts');
    assert.equal(replace.script, '/scripts/live-wrap.mjs');
    assert.deepEqual(replace.args, [
      '--id', 'abc12345', '--count', '3', '--isolated', '--element-id', 'hero',
      '--classes', 'hero,title', '--tag', 'h1', '--text', 'Exact hero copy',
    ]);
    const insert = buildDeterministicScaffoldCommand({
      id: 'abc12346',
      count: 2,
      mode: 'insert',
      insert: { position: 'before', anchor: { tag: 'section', text: 'Anchor' } },
    }, '/scripts');
    assert.equal(insert.script, '/scripts/live-insert.mjs');
    assert.deepEqual(insert.args, [
      '--id', 'abc12346', '--count', '2', '--position', 'before',
      '--tag', 'section', '--query', 'Anchor', '--text', 'Anchor',
    ]);
  });

  it('never resumes a desktop or otherwise unowned thread record', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-owner-'));
    const statePath = path.join(cwd, '.impeccable/live/codex-worker.json');
    mkdirSync(path.dirname(statePath), { recursive: true });
    writeFileSync(statePath, JSON.stringify({ owner: 'desktop', cwd, threadId: 'desktop-thread' }));
    const client = fakeClient();
    const supervisor = createSupervisor({ cwd, statePath, client });
    await supervisor.initialize();
    assert.equal(client.calls.resumeDedicatedThread.length, 0);
    assert.equal(client.calls.startDedicatedThread.length, 1);
    assert.equal(client.calls.startDedicatedThread[0].ephemeral, false);
    assert.equal(client.calls.startDedicatedThread[0].sandbox, 'read-only');
    await supervisor.shutdown();
  });

  it('resumes only a durable Live-owned worker thread', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-resume-'));
    const statePath = path.join(cwd, '.impeccable/live/codex-worker.json');
    mkdirSync(path.dirname(statePath), { recursive: true });
    writeFileSync(statePath, JSON.stringify({
      owner: CODEX_WORKER_OWNER,
      cwd,
      threadId: 'live-worker-thread',
      status: 'ready',
      threadPrimed: true,
    }));
    const client = fakeClient();
    const supervisor = createSupervisor({ cwd, statePath, client });
    await supervisor.initialize();
    assert.equal(client.calls.resumeDedicatedThread.length, 1);
    assert.equal(client.calls.resumeDedicatedThread[0].threadId, 'live-worker-thread');
    assert.equal(client.calls.startDedicatedThread.length, 0);
    assert.equal(supervisor.threadPrimed, true, 'a resumed thread must not receive the skill attachment again');
    assert.equal(JSON.parse(readFileSync(statePath, 'utf-8')).threadPrimed, true);
    await supervisor.shutdown();
  });

  it('interrupts the active dedicated turn on early Accept or Discard', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-interrupt-'));
    const client = fakeClient();
    const supervisor = createSupervisor({
      cwd,
      statePath: path.join(cwd, 'state.json'),
      client,
    });
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.model = client.models[0];
    supervisor.active = { eventId: 'generation-1', turnId: 'turn-1' };
    await supervisor.cancelActive('accept', 'generation-1');
    assert.deepEqual(client.calls.interruptTurn, [{ threadId: 'live-worker-thread', turnId: 'turn-1' }]);
    assert.equal(supervisor.canceled.has('generation-1'), true);
  });

  it('does not block deterministic Accept on the app-server interrupt round trip', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-fast-accept-'));
    const client = fakeClient();
    let releaseInterrupt;
    const interruptReleased = new Promise((resolve) => { releaseInterrupt = resolve; });
    client.interruptTurn = async (threadId, turnId) => {
      client.calls.interruptTurn.push({ threadId, turnId });
      await interruptReleased;
    };
    let acceptStarted = false;
    const events = [
      { type: 'accept', id: 'generation-1', variantId: '1' },
      { type: 'exit' },
    ];
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client,
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath: path.join(cwd, 'state.json'),
      scriptsDir: path.join(cwd, 'skill/scripts'),
      fetchEvent: async () => events.shift(),
      handleAccept: async () => {
        acceptStarted = true;
        releaseInterrupt();
        return { _acceptResult: { handled: true, carbonize: false } };
      },
    });
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.model = client.models[0];
    supervisor.active = { eventId: 'generation-1', turnId: 'turn-1', threadId: 'live-worker-thread' };

    await supervisor.run();

    assert.equal(acceptStarted, true);
    assert.equal(client.calls.interruptTurn.length >= 1, true);
  });

  it('rotates a busy thread so the next generation does not wait for the canceled tail', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-tail-rotation-'));
    const client = fakeClient();
    client.startDedicatedThread = async (params) => {
      client.calls.startDedicatedThread.push(params);
      await new Promise((resolve) => setImmediate(resolve));
      return { id: 'replacement-live-thread' };
    };
    const supervisor = createSupervisor({ cwd, statePath: path.join(cwd, 'state.json'), client });
    supervisor.model = client.models[0];
    supervisor.thread = { id: 'draining-live-thread' };
    let releaseDrainingQueue;
    let drainingQueueFinished = false;
    supervisor.queue = new Promise((resolve) => {
      releaseDrainingQueue = () => {
        drainingQueueFinished = true;
        resolve();
      };
    });
    let observedThread = null;
    supervisor.runGenerationPhase = async () => {
      observedThread = supervisor.thread.id;
    };
    supervisor.reply = async () => {};

    supervisor.rotateWorkerThread('accept');
    await supervisor.processGeneration({
      type: 'generate',
      id: 'next-generation',
      count: 1,
      scaffold: { file: 'src/App.jsx' },
    });

    assert.equal(observedThread, 'replacement-live-thread');
    assert.equal(drainingQueueFinished, false, 'the next generation must not join the canceled tail queue');
    releaseDrainingQueue();
    await new Promise((resolve) => setImmediate(resolve));
    assert.deepEqual(client.calls.archiveThread, [{ threadId: 'draining-live-thread' }]);
  });

  it('interrupts a canceled turn whose id arrives after Accept', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-late-turn-'));
    const client = fakeClient();
    const supervisor = createSupervisor({ cwd, statePath: path.join(cwd, 'state.json'), client });
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.model = client.models[0];
    supervisor.active = { eventId: 'generation-1', turnId: null };
    supervisor.canceled.add('generation-1');
    client.startTurn = async ({ onStarted }) => {
      onStarted('late-turn');
      await new Promise((resolve) => setImmediate(resolve));
      return { message: '{"files":[]}' };
    };
    await supervisor.runTurnWithReconnect({ input: 'work', outputSchema: {} });
    assert.deepEqual(client.calls.interruptTurn, [{ threadId: 'live-worker-thread', turnId: 'late-turn' }]);
  });

  it('queues carbonize cleanup onto the foreground control lane', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-carbonize-'));
    const cleanups = [];
    const order = [];
    const client = fakeClient();
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client,
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath: path.join(cwd, 'state.json'),
      scriptsDir: path.join(cwd, 'skill/scripts'),
      handleAccept: async (event) => ({
        ...event,
        _acceptResult: { handled: true, carbonize: true, file: 'src/App.jsx' },
        _completionAck: { ok: false, deferred: true },
      }),
      postCleanup: async (_base, _token, event) => { cleanups.push(event); order.push('cleanup'); },
      completeAccept: async () => { order.push('accept_ack'); },
    });
    supervisor.running = true;
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.fetchEvent = async (_base, _token, options) => {
      assert.deepEqual(options.types, CODEX_WORKER_EVENT_TYPES);
      assert.equal(options.leaseMs, CODEX_WORKER_EVENT_LEASE_MS);
      return cleanups.length === 0
        ? { type: 'accept', id: 'abc12345', variantId: '1' }
        : { type: 'exit' };
    };
    await supervisor.run();
    assert.deepEqual(cleanups, [{
      id: 'abc12345',
      sessionId: 'abc12345',
      file: 'src/App.jsx',
      variantId: '1',
      acceptResult: { handled: true, carbonize: true, file: 'src/App.jsx' },
    }]);
    assert.deepEqual(order, ['cleanup', 'accept_ack']);
  });

  it('renews but never queues the same long-running generation twice', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-duplicate-lease-'));
    const client = fakeClient();
    const supervisor = createSupervisor({
      cwd,
      statePath: path.join(cwd, 'state.json'),
      client,
    });
    supervisor.thread = { id: 'live-worker-thread' };
    let generations = 0;
    supervisor.processGeneration = async () => {
      generations += 1;
      await new Promise((resolve) => setImmediate(resolve));
    };
    const events = [
      { type: 'generate', id: 'generation-1', count: 3 },
      { type: 'generate', id: 'generation-1', count: 3 },
      { type: 'exit' },
    ];
    supervisor.fetchEvent = async () => events.shift();
    await supervisor.run();
    assert.equal(generations, 1);
    assert.equal(supervisor.queuedGenerationIds.size, 0);
  });

  it('reconnects and resumes the owned worker once after app-server loss', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-reconnect-'));
    const client = fakeClient();
    let attempts = 0;
    client.startTurn = async () => {
      attempts += 1;
      if (attempts === 1) throw new Error('app-server exited');
      return { message: '{"files":[]}' };
    };
    const supervisor = createSupervisor({
      cwd,
      statePath: path.join(cwd, 'state.json'),
      client,
    });
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.model = client.models[0];
    supervisor.active = { eventId: 'generation-1', turnId: null };
    const result = await supervisor.runTurnWithReconnect({ input: 'work', outputSchema: {} });
    assert.equal(result.answer, '{"files":[]}');
    assert.equal(client.calls.reconnect, 1);
    assert.equal(client.calls.resumeDedicatedThread.length, 1);
  });

  it('relinquishes Generate and advertises foreground fallback after permanent failure', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-fallback-'));
    const replies = [];
    const statePath = path.join(cwd, 'state.json');
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client: fakeClient(),
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath,
      scriptsDir: path.join(cwd, 'skill/scripts'),
      reply: async (_base, _token, value) => { replies.push(value); },
    });
    supervisor.running = true;
    supervisor.pollAbortController = new AbortController();

    await supervisor.handleGenerationFailure(
      { type: 'generate', id: 'recoverable-generation' },
      new Error('app-server remained unavailable after reconnect'),
    );

    assert.equal(supervisor.running, false);
    assert.equal(supervisor.pollAbortController.signal.aborted, true);
    assert.deepEqual(replies, [{
      id: 'recoverable-generation',
      type: 'retry',
      sourceEventType: 'generate',
    }]);
    const state = JSON.parse(readFileSync(statePath, 'utf-8'));
    assert.equal(state.status, 'failed');
    assert.equal(state.eventId, 'recoverable-generation');
    assert.match(state.error, /app-server remained unavailable/);
  });

  it('re-prepares once when foreground cleanup changes source during generation', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-source-race-'));
    const supervisor = createSupervisor({
      cwd,
      statePath: path.join(cwd, 'state.json'),
      client: fakeClient(),
    });
    let attempts = 0;
    supervisor.runGenerationPhaseOnce = async () => {
      attempts += 1;
      if (attempts === 1) {
        const error = new Error('publish_source_hash_mismatch');
        error.code = 'publish_source_hash_mismatch';
        throw error;
      }
    };

    await supervisor.runGenerationPhase({ id: 'source-race' }, 'first', 1);

    assert.equal(attempts, 2);
  });

  it('repairs new detector findings on the same persistent thread before publication', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-detector-repair-'));
    mkdirSync(path.join(cwd, 'src'), { recursive: true });
    const sessionId = 'detectorrepair';
    writeFileSync(path.join(cwd, 'src/App.jsx'), [
      '<main>',
      `  <div data-impeccable-variants="${sessionId}" data-impeccable-variant-count="3">`,
      `    <style data-impeccable-css="${sessionId}"></style>`,
      '    <div data-impeccable-variant="original"><h1>Original</h1></div>',
      `    {/* impeccable-variants-end ${sessionId} */}`,
      '  </div>',
      '</main>',
    ].join('\n'));
    createLiveSessionStore({ cwd, sessionId }).appendEvent({
      type: 'generate', id: sessionId, count: 3, generationEpoch: 1,
    });
    const plan = {
      identityLock: ['Preserve identity'],
      directions: [
        { variantId: 1, name: 'One', axis: 'hierarchy', intent: 'Strengthen hierarchy' },
        { variantId: 2, name: 'Two', axis: 'layout', intent: 'Recompose layout' },
        { variantId: 3, name: 'Three', axis: 'density', intent: 'Adjust density' },
      ],
    };
    const client = fakeClient();
    const turnThreadIds = [];
    const prompts = [];
    const outputSchemas = [];
    let turn = 0;
    client.startTurn = async ({ threadId, input, outputSchema, onStarted }) => {
      turn += 1;
      turnThreadIds.push(threadId);
      prompts.push(input.find((item) => item.type === 'text').text);
      outputSchemas.push(outputSchema);
      onStarted?.(`turn-${turn}`);
      return { message: JSON.stringify({
        sourceDelta: {
          variantId: 1,
          markup: `<h1>${turn === 1 ? 'Flagged' : 'Repaired'}</h1>`,
          css: '@scope ([data-impeccable-variant="1"]) { h1 { color: currentColor; } }',
        },
        plan,
        ...(turn === 1 ? {} : { detectorWaivers: [] }),
      }) };
    };
    let detectorCall = 0;
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client,
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath: path.join(cwd, '.impeccable/live/codex-worker.json'),
      scriptsDir: path.resolve('skill/scripts'),
      detectCandidate: () => {
        detectorCall += 1;
        return detectorCall === 2
          ? [{ antipattern: 'gradient-text', name: 'Gradient text', snippet: 'flagged candidate', file: 'App.jsx' }]
          : [];
      },
      publishCheckpoint: async () => {},
      publishPhase: async () => {},
    });
    supervisor.thread = { id: 'persistent-live-thread' };
    supervisor.model = client.models[0];

    await supervisor.runGenerationPhaseOnce({
      type: 'generate',
      id: sessionId,
      count: 3,
      scaffold: { file: 'src/App.jsx', styleMode: 'scoped' },
    }, 'first', 1);

    assert.deepEqual(turnThreadIds, ['persistent-live-thread', 'persistent-live-thread']);
    assert.match(prompts[1], /new Impeccable detector findings/);
    assert.equal(outputSchemas[0].properties.detectorWaivers, undefined);
    assert.equal(outputSchemas[1].properties.detectorWaivers.type, 'array');
    assert.match(readFileSync(path.join(cwd, 'src/App.jsx'), 'utf-8'), /Repaired/);
    assert.doesNotMatch(readFileSync(path.join(cwd, 'src/App.jsx'), 'utf-8'), /Flagged/);
    assert.equal(detectorCall, 3);
  });

  it('accepts only explicit narrow detector false-positive waivers', () => {
    const findings = [
      {
        antipattern: 'gradient-text',
        file: '/project/src/App.jsx',
        snippet: 'intentional campaign wordmark',
        ignoreValue: '',
      },
      {
        antipattern: 'overused-font',
        file: '/project/src/App.jsx',
        snippet: 'font-family: Inter',
        ignoreValue: 'Inter',
      },
    ];
    const resolved = resolveDetectorFindingWaivers(findings, [
      {
        rule: 'gradient-text',
        file: 'App.jsx',
        snippet: 'intentional campaign wordmark',
        ignoreValue: '',
        reason: 'The selected element is the established campaign wordmark.',
      },
      {
        rule: 'overused-font',
        file: 'App.jsx',
        snippet: '',
        ignoreValue: 'Roboto',
        reason: 'Wrong value must not waive the finding.',
      },
    ]);

    assert.deepEqual(resolved.accepted.map(({ waiver }) => waiver.rule), ['gradient-text']);
    assert.deepEqual(resolved.unresolved, [findings[1]]);
    assert.match(buildDetectorRepairPrompt('first', findings), /Fix real defects/);
    assert.match(buildDetectorRepairPrompt('first', findings), /contextually intentional or a detector false positive/);
    assert.match(buildDetectorRepairPrompt('first', findings), /do not.*persist project detector config/i);
  });

  it('publishes a candidate when the repair turn explicitly waives the remaining false positive', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-detector-waiver-'));
    mkdirSync(path.join(cwd, 'src'), { recursive: true });
    const sessionId = 'detectorwaiver';
    writeFileSync(path.join(cwd, 'src/App.jsx'), [
      '<main>',
      `  <div data-impeccable-variants="${sessionId}" data-impeccable-variant-count="1">`,
      `    <style data-impeccable-css="${sessionId}"></style>`,
      '    <div data-impeccable-variant="original"><h1>Original</h1></div>',
      `    {/* impeccable-variants-end ${sessionId} */}`,
      '  </div>',
      '</main>',
    ].join('\n'));
    createLiveSessionStore({ cwd, sessionId }).appendEvent({
      type: 'generate', id: sessionId, count: 1, generationEpoch: 1,
    });
    const finding = {
      antipattern: 'gradient-text',
      name: 'Gradient text',
      snippet: 'intentional campaign wordmark',
      ignoreValue: '',
      file: 'App.jsx',
    };
    const client = fakeClient();
    let turn = 0;
    client.startTurn = async ({ onStarted }) => {
      turn += 1;
      onStarted?.(`turn-${turn}`);
      return { message: JSON.stringify({
        sourceDelta: {
          variantId: 1,
          markup: '<h1>Intentional wordmark</h1>',
          css: '@scope ([data-impeccable-variant="1"]) { h1 { color: currentColor; } }',
        },
        ...(turn === 1 ? {} : {
          detectorWaivers: [{
            rule: 'gradient-text',
            file: 'App.jsx',
            snippet: finding.snippet,
            ignoreValue: '',
            reason: 'This selected element is the established campaign wordmark.',
          }],
        }),
      }) };
    };
    let detectorCall = 0;
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client,
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath: path.join(cwd, '.impeccable/live/codex-worker.json'),
      scriptsDir: path.resolve('skill/scripts'),
      detectCandidate: () => (++detectorCall === 1 ? [] : [finding]),
      publishCheckpoint: async () => {},
      publishPhase: async () => {},
    });
    supervisor.thread = { id: 'persistent-live-thread' };
    supervisor.model = client.models[0];

    await supervisor.runGenerationPhaseOnce({
      type: 'generate',
      id: sessionId,
      count: 1,
      scaffold: { file: 'src/App.jsx', styleMode: 'scoped' },
    }, 'first', 1);

    assert.equal(turn, 2);
    assert.equal(detectorCall, 3);
    assert.match(readFileSync(path.join(cwd, 'src/App.jsx'), 'utf-8'), /Intentional wordmark/);
    const snapshot = createLiveSessionStore({ cwd, sessionId }).getSnapshot(sessionId, { includeCompleted: true });
    assert.deepEqual(snapshot.detectorWaivers, [{
      rule: 'gradient-text',
      file: 'App.jsx',
      snippet: finding.snippet,
      ignoreValue: '',
      reason: 'This selected element is the established campaign wordmark.',
    }]);
  });

  it('resumes progressive delivery from durable variant checkpoints', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-checkpoint-resume-'));
    const phases = [];
    const replies = [];
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client: fakeClient(),
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath: path.join(cwd, 'state.json'),
      scriptsDir: path.join(cwd, 'skill/scripts'),
      reply: async (_base, _token, value) => { replies.push(value); },
    });
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.threadReady = Promise.resolve(supervisor.thread);
    supervisor.runGenerationPhase = async (_event, phase, arrivedVariants) => {
      phases.push({ phase, arrivedVariants });
    };

    const partialId = 'resume-partial';
    const store = createLiveSessionStore({ cwd, sessionId: partialId });
    store.appendEvent({ type: 'generate', id: partialId, count: 3, generationEpoch: 1 });
    store.appendEvent({ type: 'checkpoint', id: partialId, phase: 'cycling', revision: 1, arrivedVariants: 1 });
    await supervisor.processGeneration({
      type: 'generate',
      id: partialId,
      count: 3,
      generationEpoch: 1,
      scaffold: { file: 'src/App.jsx' },
    });
    assert.deepEqual(phases, [{ phase: 'remainder', arrivedVariants: 3 }]);
    assert.equal(replies.at(-1).type, 'done');

    phases.length = 0;
    const completeId = 'resume-complete';
    const completeStore = createLiveSessionStore({ cwd, sessionId: completeId });
    completeStore.appendEvent({ type: 'generate', id: completeId, count: 3, generationEpoch: 1 });
    completeStore.appendEvent({ type: 'checkpoint', id: completeId, phase: 'cycling', revision: 2, arrivedVariants: 3 });
    await supervisor.processGeneration({
      type: 'generate',
      id: completeId,
      count: 3,
      generationEpoch: 1,
      scaffold: { file: 'src/App.jsx' },
    });
    assert.deepEqual(phases, [{ phase: 'params', arrivedVariants: 3 }]);
    assert.equal(replies.at(-1).id, completeId);
    assert.equal(replies.at(-1).type, 'done');
  });

  it('archives its dedicated thread during clean Live shutdown', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-close-'));
    const client = fakeClient();
    const supervisor = createSupervisor({
      cwd,
      statePath: path.join(cwd, 'state.json'),
      client,
    });
    supervisor.thread = { id: 'live-worker-thread' };
    await supervisor.shutdown({ archive: true });
    assert.deepEqual(client.calls.archiveThread, [{ threadId: 'live-worker-thread' }]);
    assert.equal(client.calls.close, 1);
  });

  it('reports stopped rather than archived when thread archival fails', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-archive-fail-'));
    const client = fakeClient();
    client.archiveThread = async () => { throw new Error('archive unavailable'); };
    const statePath = path.join(cwd, 'state.json');
    const supervisor = createSupervisor({ cwd, statePath, client });
    supervisor.thread = { id: 'live-worker-thread' };
    await supervisor.shutdown({ archive: true });
    const state = JSON.parse(readFileSync(statePath, 'utf-8'));
    assert.equal(state.status, 'stopped');
    assert.equal(state.archived, false);
  });

  it('treats an unused thread with no persisted rollout as already archived', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-empty-thread-'));
    const statePath = path.join(cwd, 'state.json');
    const client = fakeClient();
    client.archiveThread = async () => { throw new Error('thread/archive: no rollout found for thread id empty'); };
    const supervisor = createSupervisor({ cwd, statePath, client });
    supervisor.thread = { id: 'empty' };
    await supervisor.shutdown({ archive: true });
    assert.equal(JSON.parse(readFileSync(statePath, 'utf-8')).status, 'archived');
  });

  it('publishes progressive source checkpoints only through the fenced publisher', async () => {
    const cwd = mkdtempSync(path.join(tmpdir(), 'codex-supervisor-publish-'));
    mkdirSync(path.join(cwd, 'src'), { recursive: true });
    mkdirSync(path.join(cwd, 'skill'), { recursive: true });
    writeFileSync(path.join(cwd, 'PRODUCT.md'), '# Product\nStable product context');
    writeFileSync(path.join(cwd, 'DESIGN.md'), '# Design\nStable design context');
    writeFileSync(path.join(cwd, 'skill/SKILL.md'), '# Impeccable skill');
    const sessionId = 'codexprogress';
    const original = '<main><div data-impeccable-variants="codexprogress"><style data-impeccable-css="codexprogress"></style><div data-impeccable-variant="original"><h1>Original</h1></div></div></main>';
    writeFileSync(path.join(cwd, 'src/App.jsx'), original);
    createLiveSessionStore({ cwd, sessionId }).appendEvent({
      type: 'generate',
      id: sessionId,
      count: 3,
      generationEpoch: 1,
    });
    const client = fakeClient();
    let turn = 0;
    const prompts = [];
    const turnInputs = [];
    const plan = {
      identityLock: ['Preserve copy and shared component roles'],
      directions: [
        { variantId: 1, name: 'Hierarchy', axis: 'type scale', intent: 'Strengthen hierarchy' },
        { variantId: 2, name: 'Composition', axis: 'layout', intent: 'Recompose the root' },
        { variantId: 3, name: 'Rhythm', axis: 'spacing', intent: 'Increase rhythm' },
      ],
    };
    client.startTurn = async ({ input, onStarted, onAgentMessage }) => {
      turn += 1;
      turnInputs.push(input);
      onStarted?.(`turn-${turn}`);
      const prompt = input.find((item) => item.type === 'text').text;
      prompts.push(prompt);
      const message = turn === 1
        ? JSON.stringify({
            sourceDelta: {
              variantId: 1,
              markup: '<h1>One</h1>',
              css: '@scope ([data-impeccable-variant="1"]) { h1 { color: red; } }',
            },
            plan,
          })
        : turn === 2
          ? JSON.stringify({
              sourceDeltas: [
                {
                  variantId: 2,
                  markup: '<h1>Two</h1>',
                  css: '@scope ([data-impeccable-variant="2"]) { h1 { color: green; } }',
                },
                {
                  variantId: 3,
                  markup: '<h1>Three</h1>',
                  css: '@scope ([data-impeccable-variant="3"]) { h1 { color: blue; } }',
                },
              ],
              parameterCss: '',
              paramsJson: '{"1":[],"2":[],"3":[]}',
            })
          : JSON.stringify({ parameterCss: '', paramsJson: '{"1":[],"2":[],"3":[]}' });
      await Promise.all([
        onAgentMessage?.(message),
        onAgentMessage?.(message),
      ]);
      return { message };
    };
    const replies = [];
    const checkpoints = [];
    const phases = [];
    let checkpointAttempts = 0;
    const supervisor = new CodexLiveWorkerSupervisor({
      cwd,
      base: 'http://localhost:1',
      token: 'token',
      client,
      config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
      statePath: path.join(cwd, '.impeccable/live/codex-worker.json'),
      scriptsDir: path.join(cwd, 'skill/scripts'),
      reply: async (_base, _token, value) => { replies.push(value); },
      publishCheckpoint: async (_base, _token, value) => {
        checkpointAttempts += 1;
        if (checkpointAttempts === 1) throw new Error('transient checkpoint transport failure');
        checkpoints.push(value);
      },
      publishPhase: async (_base, _token, value) => { phases.push(value); },
    });
    supervisor.thread = { id: 'live-worker-thread' };
    supervisor.model = client.models[0];

    await supervisor.processGeneration({
      type: 'generate',
      id: sessionId,
      count: 3,
      action: 'impeccable',
      scaffold: { file: 'src/App.jsx', styleMode: 'scoped' },
    });

    assert.equal(checkpoints.length, 2);
    assert.deepEqual(checkpoints.map((item) => item.arrivedVariants), [1, 3]);
    assert.deepEqual(phases.map((item) => item.phase), [
      'first_variant_generating',
      'first_variant_validating',
      'remaining_variants_generating',
      'remaining_variants_validating',
      'parameters_ready',
    ]);
    assert.equal(replies.at(-1).type, 'done');
    const publishedSource = readFileSync(path.join(cwd, 'src/App.jsx'), 'utf-8');
    assert.equal((publishedSource.match(/data-impeccable-variant="1"/g) || []).length, 2, 'selector and variant 1 remain once each');
    assert.match(publishedSource, /<h1>One<\/h1>/);
    assert.match(publishedSource, /<h1>Two<\/h1>/);
    assert.doesNotMatch(publishedSource, /Mutated/);
    const snapshot = createLiveSessionStore({ cwd, sessionId }).getSnapshot(sessionId, { includeCompleted: true });
    assert.equal(snapshot.arrivedVariants, 3);
    assert.equal(snapshot.publishedRevision, 2);
    assert.equal(snapshot.paramsPublished, true);
    assert.deepEqual(snapshot.variantPlan, plan);
    assert.equal(checkpointAttempts, 3, 'the durable first publication retries only its checkpoint');
    assert.match(prompts[1], /"name": "Composition"/);
    assert.equal(turnInputs[0].some((item) => item.type === 'skill'), true);
    assert.equal(turnInputs[1].some((item) => item.type === 'skill'), false);
    assert.equal(JSON.parse(readFileSync(path.join(cwd, '.impeccable/live/codex-worker.json'), 'utf-8')).threadPrimed, true);
    assert.equal(client.calls.startDedicatedThread.length, 0, 'both turns stay on the existing durable thread');
    assert.equal(client.calls.archiveThread.length, 0);
    assert.match(prompts[0], /Stable product context/);
    assert.match(prompts[0], /Stable design context/);
    assert.doesNotMatch(prompts[0], /<source_neighborhood>/);
    assert.doesNotMatch(prompts[1], /Stable product context|<source_neighborhood>/);
    assert.match(prompts[1], /parameterCss and paramsJson/);
  });
});

function createSupervisor({ cwd, statePath, client }) {
  return new CodexLiveWorkerSupervisor({
    cwd,
    base: 'http://localhost:1',
    token: 'token',
    client,
    config: { model: null, effort: 'low', delivery: 'progressive', maxArtifactBytes: 2_000_000 },
    statePath,
    scriptsDir: path.join(cwd, 'skill/scripts'),
  });
}

function fakeClient() {
  const calls = {
    connect: 0,
    listModels: 0,
    startDedicatedThread: [],
    resumeDedicatedThread: [],
    reconnect: 0,
    interruptTurn: [],
    archiveThread: [],
    close: 0,
  };
  const models = [{
    id: 'gpt-5.3-codex-spark',
    model: 'gpt-5.3-codex-spark',
    supportedReasoningEfforts: [{ reasoningEffort: 'low' }],
  }];
  return {
    calls,
    models,
    async connect() { calls.connect += 1; },
    async listModels() { calls.listModels += 1; return models; },
    async startDedicatedThread(params) { calls.startDedicatedThread.push(params); return { id: 'new-live-thread' }; },
    async resumeDedicatedThread(threadId, params) {
      calls.resumeDedicatedThread.push({ threadId, ...params });
      return { id: threadId };
    },
    async reconnect({ threadId, resumeParams }) {
      calls.reconnect += 1;
      calls.resumeDedicatedThread.push({ threadId, ...resumeParams });
      return { id: threadId };
    },
    async startTurn() { return { message: 'READY' }; },
    async interruptTurn(threadId, turnId) { calls.interruptTurn.push({ threadId, turnId }); },
    async archiveThread(threadId) { calls.archiveThread.push({ threadId }); },
    async close() { calls.close += 1; },
  };
}