import assert from 'node:assert/strict';
import test from 'node:test';
import WebSocket from 'ws';
import { InMemoryRepository } from '@platform-modules/chatgpt-orchestrator-persistence';
import { OrchestratorService } from '@platform-modules/chatgpt-orchestrator-core';
import { PROTOCOL_VERSION, type ExecutorCommand, type ExecutorServerFrame } from '@platform-modules/chatgpt-orchestrator-protocol';
import { ExecutorBridge } from '../src/executor-bridge.js';
import { workerBootstrap } from '../src/bootstrap.js';

async function nextFrames(socket: WebSocket, count: number): Promise<ExecutorServerFrame[]> {
  return new Promise((resolve, reject) => {
    const frames: ExecutorServerFrame[] = [];
    const onMessage = (data: WebSocket.RawData) => {
      frames.push(JSON.parse(data.toString()) as ExecutorServerFrame);
      if (frames.length >= count) {
        cleanup();
        resolve(frames);
      }
    };
    const onError = (error: Error) => { cleanup(); reject(error); };
    const cleanup = () => { socket.off('message', onMessage); socket.off('error', onError); };
    socket.on('message', onMessage);
    socket.once('error', onError);
  });
}

async function nextFrameOfType<T extends ExecutorServerFrame['type']>(socket: WebSocket, type: T): Promise<Extract<ExecutorServerFrame, { type: T }>> {
  return new Promise((resolve, reject) => {
    const onMessage = (data: WebSocket.RawData) => {
      const frame = JSON.parse(data.toString()) as ExecutorServerFrame;
      if (frame.type !== type) return;
      cleanup();
      resolve(frame as Extract<ExecutorServerFrame, { type: T }>);
    };
    const onError = (error: Error) => { cleanup(); reject(error); };
    const cleanup = () => { socket.off('message', onMessage); socket.off('error', onError); };
    socket.on('message', onMessage);
    socket.once('error', onError);
  });
}

async function open(url: string): Promise<WebSocket> {
  const socket = new WebSocket(url);
  await new Promise<void>((resolve, reject) => {
    socket.once('open', () => resolve());
    socket.once('error', reject);
  });
  return socket;
}

test('bridge authenticates, dispatches, acks, records state, and completes command', async () => {
  const repository = new InMemoryRepository();
  const service = new OrchestratorService(repository);
  const { run, rootWorker } = await service.createRun({ title: 'bridge' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'child',
    assignment: { objective: 'work' },
  });
  const command: ExecutorCommand = {
    protocolVersion: PROTOCOL_VERSION,
    commandId: `cmd_${worker.workerId.slice(4)}`,
    type: 'conversation.create',
    workerId: worker.workerId,
    idempotencyKey: `create:${worker.workerId}`,
    deadline: new Date(Date.now() + 60_000).toISOString(),
    bootstrap: 'Attach worker for executor bridge test.',
  };
  await service.enqueueExecutorCommand(command);

  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token: 'a'.repeat(32), heartbeatIntervalMs: 1000, staleAfterMs: 5000 });
  await bridge.start();
  const socket = await open(bridge.address);
  try {
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'test-firefox',
      token: 'a'.repeat(32),
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    const [ready, dispatched] = await nextFrames(socket, 2);
    assert.equal(ready?.type, 'executor.ready');
    assert.equal(dispatched?.type, 'executor.command');
    if (!dispatched || dispatched.type !== 'executor.command') throw new Error('expected command frame');
    assert.equal(dispatched.command.commandId, command.commandId);

    socket.send(JSON.stringify({ protocolVersion: PROTOCOL_VERSION, type: 'command.ack', executorId: 'test-firefox', commandId: command.commandId }));
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'conversation.state',
      executorId: 'test-firefox',
      workerId: worker.workerId,
      state: 'idle',
      tabId: 42,
      conversationId: 'c-test',
      url: 'https://chatgpt.com/c/c-test',
    }));
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'command.result',
      executorId: 'test-firefox',
      commandId: command.commandId,
      success: true,
      result: { tabId: 42, conversationId: 'c-test' },
    }));

    await new Promise((resolve) => setTimeout(resolve, 30));
    const stored = await repository.read((state) => state.executorCommands[command.commandId]);
    assert.equal(stored?.status, 'completed');
    const binding = await service.getConversationBinding(worker.workerId);
    assert.equal(binding?.tabId, 42);
    assert.equal(binding?.conversationId, 'c-test');
  } finally {
    socket.close();
    await bridge.close();
  }
});

test('bridge rejects a wrong executor token before dispatch', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token: 'b'.repeat(32) });
  await bridge.start();
  const socket = await open(bridge.address);
  try {
    const closed = new Promise<number>((resolve) => socket.once('close', (code) => resolve(code)));
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'test-firefox',
      token: 'c'.repeat(32),
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    assert.equal(await closed, 1008);
  } finally {
    await bridge.close();
  }
});

test('duplicate authenticated browser contexts stay connected as standby and promote without reconnect war', async () => {
  const repository = new InMemoryRepository();
  const service = new OrchestratorService(repository);
  const { run, rootWorker } = await service.createRun({ title: 'duplicate browser contexts' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'standby',
    assignment: { objective: 'survive duplicate MV3 contexts' },
  });
  const command: ExecutorCommand = {
    protocolVersion: PROTOCOL_VERSION,
    commandId: `cmd_${worker.workerId.slice(4)}`,
    type: 'conversation.create',
    workerId: worker.workerId,
    idempotencyKey: `standby:${worker.workerId}`,
    deadline: new Date(Date.now() + 60_000).toISOString(),
    bootstrap: 'Attach worker for duplicate browser-context test.',
  };
  await service.enqueueExecutorCommand(command);

  const token = 's'.repeat(32);
  const executorId = 'duplicate-firefox';
  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token });
  await bridge.start();
  try {
    const first = await open(bridge.address);
    first.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId,
      token,
      browser: 'firefox',
      extensionVersion: '0.1.1',
    }));
    const firstFrames = await nextFrames(first, 2);
    assert.equal(firstFrames[0]?.type, 'executor.ready');
    assert.equal(firstFrames[1]?.type, 'executor.command');
    first.send(JSON.stringify({ protocolVersion: PROTOCOL_VERSION, type: 'command.ack', executorId, commandId: command.commandId }));

    const second = await open(bridge.address);
    second.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId,
      token,
      browser: 'firefox',
      extensionVersion: '0.1.1',
    }));
    const [standbyReady] = await nextFrames(second, 1);
    assert.equal(standbyReady?.type, 'executor.ready');
    assert.equal(first.readyState, WebSocket.OPEN);
    assert.equal(second.readyState, WebSocket.OPEN);

    await new Promise<void>((resolve, reject) => {
      const timer = setTimeout(() => { second.off('message', onMessage); resolve(); }, 40);
      const onMessage = (data: WebSocket.RawData) => {
        clearTimeout(timer);
        second.off('message', onMessage);
        reject(new Error(`standby unexpectedly received frame: ${data.toString()}`));
      };
      second.on('message', onMessage);
    });

    const firstClosed = new Promise<void>((resolve) => first.once('close', () => resolve()));
    first.close(1000, 'promote standby');
    await firstClosed;
    const promotedFrames = await nextFrames(second, 2);
    assert.equal(promotedFrames[0]?.type, 'executor.ready');
    assert.equal(promotedFrames[1]?.type, 'executor.command');
    if (promotedFrames[1]?.type !== 'executor.command') throw new Error('expected command replay after standby promotion');
    assert.equal(promotedFrames[1].command.commandId, command.commandId);

    second.send(JSON.stringify({ protocolVersion: PROTOCOL_VERSION, type: 'command.ack', executorId, commandId: command.commandId }));
    second.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'command.result',
      executorId,
      commandId: command.commandId,
      success: true,
      result: { promoted: true },
    }));
    await new Promise((resolve) => setTimeout(resolve, 20));
    const stored = await repository.read((state) => state.executorCommands[command.commandId]);
    assert.equal(stored?.status, 'completed');
    second.close();
  } finally {
    await bridge.close();
  }
});

test('incomplete command replays once after executor reconnect and then becomes terminal', async () => {
  const repository = new InMemoryRepository();
  const service = new OrchestratorService(repository);
  const { run, rootWorker } = await service.createRun({ title: 'reconnect' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'replay',
    assignment: { objective: 'replay safely' },
  });
  const command: ExecutorCommand = {
    protocolVersion: PROTOCOL_VERSION,
    commandId: `cmd_${worker.workerId.slice(4)}`,
    type: 'conversation.create',
    workerId: worker.workerId,
    idempotencyKey: `replay:${worker.workerId}`,
    deadline: new Date(Date.now() + 60_000).toISOString(),
    bootstrap: 'Attach worker for executor bridge test.',
  };
  await service.enqueueExecutorCommand(command);

  const token = 'd'.repeat(32);
  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token });
  await bridge.start();
  try {
    const first = await open(bridge.address);
    first.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'reconnect-firefox',
      token,
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    const firstFrames = await nextFrames(first, 2);
    assert.equal(firstFrames[1]?.type, 'executor.command');
    first.send(JSON.stringify({ protocolVersion: PROTOCOL_VERSION, type: 'command.ack', executorId: 'reconnect-firefox', commandId: command.commandId }));
    const firstClosed = new Promise<void>((resolve) => first.once('close', () => resolve()));
    first.close(1000, 'test disconnect');
    await firstClosed;

    const second = await open(bridge.address);
    second.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'reconnect-firefox',
      token,
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    const secondFrames = await nextFrames(second, 2);
    assert.equal(secondFrames[1]?.type, 'executor.command');
    if (secondFrames[1]?.type !== 'executor.command') throw new Error('expected replayed command');
    assert.equal(secondFrames[1].command.commandId, command.commandId);
    second.send(JSON.stringify({ protocolVersion: PROTOCOL_VERSION, type: 'command.ack', executorId: 'reconnect-firefox', commandId: command.commandId }));
    second.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'command.result',
      executorId: 'reconnect-firefox',
      commandId: command.commandId,
      success: true,
      result: { replayed: true },
    }));
    await new Promise((resolve) => setTimeout(resolve, 20));
    const stored = await repository.read((state) => state.executorCommands[command.commandId]);
    assert.equal(stored?.status, 'completed');
    assert.equal(stored?.attempts, 2);
    second.close();
  } finally {
    await bridge.close();
  }
});


test('authenticated operator actions return authoritative results and continuity views', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const { run, rootWorker } = await service.createRun({ title: 'operator controls' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'operator-child',
    assignment: { objective: 'exercise operator controls' },
  });
  const token = 'e'.repeat(32);
  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token });
  await bridge.start();
  const socket = await open(bridge.address);
  try {
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'operator-firefox',
      token,
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    await nextFrameOfType(socket, 'executor.ready');

    const pauseResultPromise = nextFrameOfType(socket, 'operator.result');
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'operator.action',
      executorId: 'operator-firefox',
      requestId: 'pause-1',
      workerId: worker.workerId,
      action: 'pause',
    }));
    const pauseResult = await pauseResultPromise;
    assert.equal(pauseResult.success, true);
    assert.equal(pauseResult.requestId, 'pause-1');
    assert.equal(pauseResult.view?.disposition, 'paused');
    assert.equal(pauseResult.view?.runId, run.runId);
    assert.equal(pauseResult.view?.runTitle, 'operator controls');
    assert.equal(pauseResult.view?.workerName, '/root/operator-child');
    assert.equal(pauseResult.view?.policySource, 'run');
    assert.equal(pauseResult.view?.workerPolicyOverride, false);
    assert.equal(pauseResult.view?.pauseReason, 'user');
    assert.equal(typeof pauseResult.view?.noProgressAttempts, 'number');
    assert.ok(Array.isArray(pauseResult.view?.recentEvents));
    assert.equal((await service.getWorkerContinuity(worker.workerId)).continuity.disposition, 'paused');

    const policyResultPromise = nextFrameOfType(socket, 'operator.result');
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'operator.action',
      executorId: 'operator-firefox',
      requestId: 'auto-1',
      workerId: worker.workerId,
      action: 'set_auto_resume',
      enabled: true,
    }));
    const policyResult = await policyResultPromise;
    assert.equal(policyResult.success, true);
    assert.equal(policyResult.view?.policyMode, 'auto');
    assert.equal(policyResult.view?.policySource, 'worker');
    assert.equal(policyResult.view?.workerPolicyOverride, true);

    const runPolicyResultPromise = nextFrameOfType(socket, 'operator.result');
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'operator.action',
      executorId: 'operator-firefox',
      requestId: 'run-auto-1',
      runId: run.runId,
      action: 'set_run_auto_resume',
      enabled: true,
    }));
    const runPolicyResult = await runPolicyResultPromise;
    assert.equal(runPolicyResult.success, true);
    assert.equal(runPolicyResult.runId, run.runId);
    assert.ok(runPolicyResult.views?.some((view) => view.workerId === worker.workerId));
    assert.equal((await service.getRunContinuationPolicy(run.runId)).mode, 'auto');

    const resumeRunResultPromise = nextFrameOfType(socket, 'operator.result');
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'operator.action',
      executorId: 'operator-firefox',
      requestId: 'resume-run-1',
      runId: run.runId,
      action: 'resume_run',
    }));
    const resumeRunResult = await resumeRunResultPromise;
    assert.equal(resumeRunResult.success, true);
    assert.notEqual((await service.getWorkerContinuity(worker.workerId)).continuity.disposition, 'paused');
  } finally {
    socket.close();
    await bridge.close();
  }
});

test('idle assistant blocked output is parsed fail-safe and reconciled into a durable blocker', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const { run, rootWorker } = await service.createRun({ title: 'blocked output' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'blocked',
    assignment: { objective: 'work until externally blocked' },
  });
  const attempt = (await service.listExecutionAttempts(worker.workerId))[0]!;
  const command: ExecutorCommand = {
    protocolVersion: PROTOCOL_VERSION,
    commandId: `cmd_${attempt.executionId.slice(4)}`,
    type: 'conversation.create',
    workerId: worker.workerId,
    idempotencyKey: `conversation.execution:${attempt.executionId}`,
    deadline: new Date(Date.now() + 60_000).toISOString(),
    bootstrap: workerBootstrap(worker.workerId),
  };
  await service.enqueueExecutorCommand(command);

  const token = 'f'.repeat(32);
  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token });
  await bridge.start();
  const socket = await open(bridge.address);
  try {
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'blocked-firefox',
      token,
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    await nextFrames(socket, 2);
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'command.result',
      executorId: 'blocked-firefox',
      commandId: command.commandId,
      success: true,
      result: {},
    }));
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'conversation.state',
      executorId: 'blocked-firefox',
      workerId: worker.workerId,
      state: 'generating',
      conversationId: 'c-blocked',
    }));
    await new Promise((resolve) => setTimeout(resolve, 10));

    const continuityPromise = nextFrameOfType(socket, 'worker.continuity');
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'conversation.state',
      executorId: 'blocked-firefox',
      workerId: worker.workerId,
      state: 'idle',
      conversationId: 'c-blocked',
      assistantOutputTail: 'Result so far.\n<continuation-state>\nstatus: blocked\nsummary: waiting for external artifact\ncurrent: implementation cannot proceed\nnext: provide the artifact\n</continuation-state>',
      assistantOutputTruncated: false,
    }));
    const continuity = await continuityPromise;
    assert.equal(continuity.view.disposition, 'awaiting_dependency');
    assert.equal(continuity.view.needsUser, true);
    assert.equal((await service.getWorkerContinuity(worker.workerId)).continuity.disposition, 'awaiting_dependency');
  } finally {
    socket.close();
    await bridge.close();
  }
});

test('message delivery timeout records the timeout and dispatches a fresh recovery send instead of Retry', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const { run, rootWorker } = await service.createRun({ title: 'delivery timeout' });
  await service.setRunContinuationPolicy(run.runId, { mode: 'auto', idleGraceMs: 0 });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'timeout',
    assignment: { objective: 'recover delivery safely' },
  });
  const attempt = (await service.listExecutionAttempts(worker.workerId))[0]!;
  const command: ExecutorCommand = {
    protocolVersion: PROTOCOL_VERSION,
    commandId: `cmd_${attempt.executionId.slice(4)}`,
    type: 'conversation.create',
    workerId: worker.workerId,
    idempotencyKey: `conversation.execution:${attempt.executionId}`,
    deadline: new Date(Date.now() + 60_000).toISOString(),
    bootstrap: workerBootstrap(worker.workerId),
  };
  await service.enqueueExecutorCommand(command);

  const token = 'g'.repeat(32);
  const bridge = new ExecutorBridge(service, { host: '127.0.0.1', port: 0, token });
  await bridge.start();
  const socket = await open(bridge.address);
  try {
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'executor.hello',
      executorId: 'timeout-firefox',
      token,
      browser: 'firefox',
      extensionVersion: '0.1.0',
    }));
    await nextFrames(socket, 2);
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'command.result',
      executorId: 'timeout-firefox',
      commandId: command.commandId,
      success: true,
      result: {},
    }));
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'conversation.state',
      executorId: 'timeout-firefox',
      workerId: worker.workerId,
      state: 'generating',
      conversationId: 'c-timeout',
    }));
    await new Promise((resolve) => setTimeout(resolve, 10));

    const recoveryPromise = nextFrameOfType(socket, 'executor.command');
    socket.send(JSON.stringify({
      protocolVersion: PROTOCOL_VERSION,
      type: 'conversation.state',
      executorId: 'timeout-firefox',
      workerId: worker.workerId,
      state: 'error',
      conversationId: 'c-timeout',
      error: 'MESSAGE_DELIVERY_TIMEOUT: ChatGPT did not accept the message',
    }));
    const recovery = await recoveryPromise;
    assert.equal(recovery.command.type, 'conversation.send');
    if (recovery.command.type !== 'conversation.send') throw new Error('expected fresh recovery send');
    assert.match(recovery.command.prompt, /Recover authoritative state using the ChatGPT Orchestrator MCP/);
    assert.doesNotMatch(recovery.command.prompt, /^continue$/i);
    assert.doesNotMatch(recovery.command.prompt, /Retry/i);
    const attempts = await service.listExecutionAttempts(worker.workerId);
    assert.equal(attempts[0]?.outcome, 'message_delivery_timeout');
    assert.equal(attempts.at(-1)?.reason, 'recovery');
  } finally {
    socket.close();
    await bridge.close();
  }
});