/** Disposable real-Codex crash gate. Leaves the existing Codex login unchanged.
 * HTTP uses the production Supen RPC handler, on an ephemeral loopback port.
 */
import assert from 'node:assert/strict';
import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import http from 'node:http';
import { spawn, type ChildProcess } from 'node:child_process';
import { fileURLToPath } from 'node:url';

const mode = process.argv[2];
const waitFor = async <T>(read: () => Promise<T | undefined>, label: string): Promise<T> => {
  const deadline = Date.now() + 120_000;
  while (Date.now() < deadline) {
    const value = await read();
    if (value !== undefined) return value;
    await new Promise((resolve) => setTimeout(resolve, 50));
  }
  throw new Error(`Timed out: ${label}`);
};

if (mode) {
  const { getSharedCodexAppServerHost } = await import('../src/core/codex-app-server-host.js');
  const host = getSharedCodexAppServerHost();
  await host.ensureInitialized({ cwd: process.env.SUPEN_RECOVERY_CWD! });
  process.send?.({ runtimePid: (host as any).child.pid });
  const request = host.request.bind(host);
  host.request = async (method, params) => {
    if (mode === 'queued' && method === 'thread/resume') {
      process.send?.({ checkpoint: true });
      return new Promise(() => {});
    }
    const result = await request(method, params);
    if (mode === 'unknown' && method === 'turn/start') return new Promise(() => {});
    if (mode === 'running' && method === 'turn/start') setTimeout(() => process.send?.({ checkpoint: true }), 25);
    return result;
  };
  if (mode === 'unknown') {
    host.subscribeNotifications((event) => {
      if (event.method === 'turn/completed') process.send?.({ checkpoint: true });
    });
  }
  if (mode === 'completed') {
    const open = host.openNotificationChannel.bind(host);
    host.openNotificationChannel = (threadId) => {
      const channel = open(threadId);
      return {
        close: () => channel.close(),
        async next(...args) {
          const event = await channel.next(...args);
          if (event?.method === 'turn/completed') {
            process.send?.({ checkpoint: true });
            return new Promise(() => {});
          }
          return event;
        },
      };
    };
  }
  const { handleRpcRoutes } = await import('../src/http/routes/rpc.js');
  const server = http.createServer((req, res) => {
    void handleRpcRoutes(req, res, '/api/computers/{computer_id}/agents/codex/rpc', req.method || '', null,
      () => { throw new Error('Unexpected legacy dispatch'); }).catch((error) => {
      res.statusCode = 500;
      res.end(String(error));
    });
  });
  server.listen(0, '127.0.0.1', () => process.send?.({ port: (server.address() as any).port }));
  process.on('message', (message) => {
    if (message === 'stop') {
      host.close();
      server.close(() => process.exit(0));
    }
  });
} else {
  const root = fs.mkdtempSync(path.join(os.tmpdir(), 'supen-followup-crash-'));
  process.env.SUPEN_HOME = path.join(root, 'supen');
  process.env.SUPEN_RECOVERY_CWD = path.join(root, 'workspace');
  fs.mkdirSync(process.env.SUPEN_RECOVERY_CWD);
  const { getSharedCodexAppServerHost, terminateAppServerProcessTreeByPid } = await import('../src/core/codex-app-server-host.js');
  const { CodexAppServerDriver } = await import('../src/agent-sdk/drivers/codex-app-server-driver.js');
  const host = getSharedCodexAppServerHost();
  let threadId = '';
  let worker: ChildProcess | undefined;
  let runtimePid: number | undefined;
  let workerEvents: Array<Record<string, any>> = [];
  let workerLog = '';
  const launch = async (checkpoint: string) => {
    workerEvents = [];
    workerLog = '';
    runtimePid = undefined;
    worker = spawn(process.execPath, ['--import', 'tsx', fileURLToPath(import.meta.url), checkpoint], {
      env: process.env, stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
    });
    worker.stdout?.on('data', (data) => { workerLog += String(data); });
    worker.stderr?.on('data', (data) => { workerLog += String(data); });
    worker.on('message', (message: any) => {
      workerEvents.push(message);
      if (message.runtimePid) runtimePid = message.runtimePid;
    });
    return waitFor(async () => {
      if (worker?.exitCode !== null) throw new Error(`Worker exited: ${workerLog}`);
      return workerEvents.find((event) => event.port)?.port as number | undefined;
    }, 'RPC worker listen');
  };
  const stop = async (crash: boolean) => {
    const current = worker;
    if (!current) return;
    const exited = new Promise<void>((resolve) => {
      if (current.exitCode !== null || current.signalCode !== null) resolve();
      else current.once('exit', () => resolve());
    });
    if (crash) {
      current.kill('SIGKILL');
      terminateAppServerProcessTreeByPid(runtimePid, 'SIGKILL');
    } else {
      current.send('stop');
    }
    await exited;
    worker = undefined;
    runtimePid = undefined;
  };
  const rpc = async (port: number, method: string, params: Record<string, unknown> = {}) => {
    const response = await fetch(`http://127.0.0.1:${port}/rpc`, {
      method: 'POST', headers: { 'content-type': 'application/json' },
      body: JSON.stringify({ jsonrpc: '2.0', id: 1, method, params: { agentId: 'codex', threadId, ...params } }),
    });
    assert.equal(response.status, 200);
    return response.json() as Promise<any>;
  };
  const history = async () => {
    await host.ensureInitialized({ cwd: process.env.SUPEN_RECOVERY_CWD! });
    const response = await host.request('thread/read', { threadId, includeTurns: true });
    host.close();
    return (response.result?.thread as any).turns as Array<any>;
  };
  try {
    await host.ensureInitialized({ cwd: process.env.SUPEN_RECOVERY_CWD! });
    const empty = await host.request('thread/start', {
      cwd: process.env.SUPEN_RECOVERY_CWD!, approvalPolicy: 'never', sandbox: 'read-only',
    });
    threadId = (empty.result?.thread as any).id;
    assert(threadId);
    const seed = await new CodexAppServerDriver(() => host).startThread({
      agentId: 'codex', threadId: 'seed', turnId: 'seed', cwd: process.env.SUPEN_RECOVERY_CWD!,
      resume: threadId, resumeRequired: true,
      permissionMode: 'read-only', networkAccess: false,
    });
    await seed.send('Reply exactly SEED. Do not use tools.');
    for await (const event of seed.stream()) {
      if ((event as any).thread_id) threadId = (event as any).thread_id;
    }
    assert(threadId);
    console.log('PASS: first turn on a native-created, unmaterialized thread');
    host.close();
    let expectedTurns = (await history()).length;
    for (const checkpoint of ['queued', 'completed', 'unknown', 'running']) {
      const marker = `RECOVERY-${checkpoint}-${Date.now()}`;
      const port = await launch(checkpoint);
      const started = await rpc(port, 'turn/start', {
        input: [{ type: 'text', text: `Reply exactly ${marker}. Do not use tools.` }],
        cwd: process.env.SUPEN_RECOVERY_CWD, permissionMode: 'read-only',
      });
      assert.equal(started.result?.turn.status, 'accepted', JSON.stringify(started));
      await waitFor(async () => workerEvents.some((event) => event.checkpoint) ? true : undefined, checkpoint);
      await stop(true);
      const durable = JSON.parse(fs.readFileSync(path.join(process.env.SUPEN_HOME!, 'runtime', 'codex-followups.json'), 'utf8'));
      assert.equal(durable.length, 1);
      assert.equal(durable[0].state, checkpoint === 'queued' ? 'queued' : 'dispatching');
      assert.equal(!!durable[0].nativeTurnId, checkpoint === 'completed' || checkpoint === 'running');
      const resumedPort = await launch('restore');
      if (checkpoint === 'unknown' || checkpoint === 'running') {
        await waitFor(async () => {
          const list = await rpc(resumedPort, 'thread/followups/list');
          return list.result?.followups[0]?.state === 'blocked' ? true : undefined;
        }, 'unknown receipt blocks without replay');
        const refused = await rpc(resumedPort, 'thread/followups/discard', { turnId: durable[0].turnId });
        assert(refused.error);
        const discarded = await rpc(resumedPort, 'thread/followups/discard', {
          turnId: durable[0].turnId, acknowledgeUncertain: true,
        });
        assert.deepEqual(discarded.result?.followups, []);
      } else {
        await waitFor(async () => {
          const list = await rpc(resumedPort, 'thread/followups/list');
          assert(!list.result?.followups.some((item: any) => item.state === 'blocked'), JSON.stringify(list));
          return list.result?.followups.length === 0 ? true : undefined;
        }, 'recovered delivery settles');
      }
      await stop(false);
      let turns = await history();
      assert.equal(turns.length, ++expectedTurns, `${checkpoint}: duplicate or missing turn`);
      if (checkpoint !== 'running') {
        assert(turns.some((turn) => turn.items.some((item: any) => item.type === 'agentMessage' && item.text.includes(marker))),
          `${checkpoint}: missing real model output`);
      } else {
        const followupPort = await launch('restore');
        const followup = await rpc(followupPort, 'turn/start', {
          input: [{ type: 'text', text: `Continue by replying exactly ${marker}. Do not use tools.` }],
          cwd: process.env.SUPEN_RECOVERY_CWD, permissionMode: 'read-only',
        });
        assert(followup.result, JSON.stringify(followup));
        await waitFor(async () => {
          const list = await rpc(followupPort, 'thread/followups/list');
          return list.result?.followups.length === 0 ? true : undefined;
        }, 'explicit continuation after interrupted runner');
        await stop(false);
        turns = await history();
        assert.equal(turns.length, ++expectedTurns);
        assert(turns.some((turn) => turn.items.some((item: any) => item.type === 'agentMessage' && item.text.includes(marker))));
      }
      console.log(JSON.stringify({ checkpoint, threadId, httpAcceptanceDurable: true, noDuplicateNativeTurn: true }));
    }
  } finally {
    await stop(true);
    try {
      if (threadId) {
        await host.ensureInitialized({ cwd: process.env.SUPEN_RECOVERY_CWD! });
        await host.request('thread/archive', { threadId });
        console.log(JSON.stringify({ threadId, archived: true }));
      }
    } finally {
      host.close();
      fs.rmSync(root, { recursive: true, force: true });
    }
  }
}