/**
 * Subagents v2 — Codex harness (Davis protocol).
 *
 * Spawns subagents via `codex app-server --stdio` using the real
 * JSON-RPC / JSONL protocol. Does NOT emit subagent:spawned — the manager owns it.
 *
 * Protocol flow:
 *   1. initialize → initialized notification
 *   2. thread/start { cwd, approvalPolicy: 'never', sandbox, model }
 *   3. turn/start { threadId, input: [{type:'text',text,text_elements:[]}], effort }
 *   4. Stream notifications: turn/started, item/started, item/completed,
 *      item/agentMessage/delta, turn/completed, thread/tokenUsage/updated
 *   5. Approval requests are rejected
 *   6. turn/interrupt for cancellation
 *
 * Lifecycle: turn/completed resolves a dedicated turn promise; then we
 * terminate the process. We do NOT wait forever for process exit.
 *
 * Cleanup:
 *   Windows: taskkill /T /F on the process tree
 *   POSIX: SIGTERM on detached process group, then SIGKILL after grace
 */
import { type ChildProcess } from "node:child_process";
import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import * as readline from "node:readline";
import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
import type {
  SubagentHarness,
  SubagentSpawnParams,
  SubagentRecord,
  SubagentEvent,
  CodexSandbox,
} from "./types.ts";
import {
  resolveCodexInvocation,
  resolveCodexSandbox,
  resolveCodexMaxEffort,
  resolveModel,
  resolveThinking,
} from "./config.ts";
import { normalizeToolEvent } from "./protocol.ts";
import { spawnDetached, terminateProcessTree } from "./process-helpers.ts";

// ── Constants ─────────────────────────────────────────────────────────────────

const TURN_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes per turn

// ── Davis protocol types ─────────────────────────────────────────────────────

interface JsonRpcRequest {
  method: string;
  id: number;
  params?: Record<string, unknown>;
}

interface JsonRpcResponse {
  id?: number;
  result?: unknown;
  error?: { code: number; message: string };
}

interface JsonRpcNotification {
  method: string;
  params?: Record<string, unknown>;
}

type JsonRpcMessage = JsonRpcRequest | JsonRpcResponse | JsonRpcNotification;

interface PendingRequest {
  resolve: (response: JsonRpcResponse) => void;
  reject: (err: Error) => void;
  timer: ReturnType<typeof setTimeout>;
}

// ── Harness ──────────────────────────────────────────────────────────────────

export class CodexSubagentHarness implements SubagentHarness {
  readonly kind = "codex" as const;
  private activeProcesses = new Map<string, ChildProcess>();

  async spawn(
    id: string,
    params: SubagentSpawnParams,
    _ctx: ExtensionContext,
    onEvent: (event: SubagentEvent) => void,
    signal?: AbortSignal,
  ): Promise<void> {
    const cwd = params.cwd ?? _ctx.cwd;
    const codexInvocation = resolveCodexInvocation();
    const sandbox = resolveCodexSandbox(params.codexSandbox);
    const maxEffort = resolveCodexMaxEffort();
    const effort = clampEffort(
      resolveThinking("codex", params.thinking) ?? maxEffort,
      maxEffort,
    );
    // Omit the model when neither the call nor subagents.json specifies one;
    // Codex then uses its own authenticated/configured default.
    const model = resolveModel("codex", params.model);

    const startedAt = Date.now();

    const record: SubagentRecord = {
      id,
      harness: "codex",
      status: "running",
      label: params.label ?? `codex-${id.slice(0, 8)}`,
      task: params.task,
      cwd,
      model: model ?? "codex-default",
      thinking: effort,
      startedAt,
      completedAt: null,
      toolCount: 0,
      output: "",
      followUpDelivered: false,
      usage: {
        input: 0,
        output: 0,
        cacheRead: 0,
        cacheWrite: 0,
        cost: 0,
        turns: 0,
      },
      recentTools: [],
    };

    // Manager emits spawned; harness emits started
    onEvent({
      type: "subagent:started",
      id,
      harness: "codex",
      timestamp: startedAt,
      record,
    });

    let proc: ChildProcess | undefined;
    let codexSqliteHome: string | undefined;
    let turnTimer: ReturnType<typeof setTimeout> | undefined;
    let abortHandler: (() => void) | undefined;

    try {
      // Each app-server gets an isolated SQLite runtime. Current Codex builds
      // can fail startup when multiple clients share ~/.codex SQLite files;
      // CODEX_SQLITE_HOME keeps auth/config in the normal CODEX_HOME while
      // avoiding that lock. The thread itself is ephemeral.
      codexSqliteHome = await mkdtemp(join(tmpdir(), "pi-codex-subagent-"));

      // Spawn codex app-server --stdio. On Windows, resolveCodexInvocation
      // turns npm's codex.cmd wrapper into `node .../codex.js` so stdin/stdout
      // remain real pipes instead of passing through cmd.exe.
      proc = spawnDetached(
        codexInvocation.command,
        [...codexInvocation.argsPrefix, "app-server", "--stdio"],
        { cwd, env: { CODEX_SQLITE_HOME: codexSqliteHome } },
      );
      this.activeProcesses.set(id, proc);

      // Set up JSONL reader
      const rl = readline.createInterface({
        input: proc.stdout!,
        crlfDelay: Infinity,
      });

      let nextId = 0;
      const pending = new Map<number, PendingRequest>();
      let outputText = "";
      let reasoningText = "";
      let threadId: string | undefined;
      let turnId: string | undefined;

      // ── Turn promise ────────────────────────────────────────────────────
      // Resolved when turn/completed notification arrives, or on error.
      let turnSettled = false;
      let turnResolve: (result: { output: string; status: string }) => void;
      let turnReject: (err: Error) => void;
      const turnPromise = new Promise<{ output: string; status: string }>(
        (resolve, reject) => {
          turnResolve = (result) => {
            if (turnSettled) return;
            turnSettled = true;
            resolve(result);
          };
          turnReject = (error) => {
            if (turnSettled) return;
            turnSettled = true;
            reject(error);
          };
        },
      );
      // The process can fail during handshake, before the turn race attaches.
      // Attach a handler now to prevent an unhandled-rejection crash; the
      // original promise remains rejected for the later race.
      void turnPromise.catch(() => {});

      // ── RPC helpers ─────────────────────────────────────────────────────

      const request = (
        method: string,
        rpcParams?: Record<string, unknown>,
        timeoutMs = 30_000,
      ): Promise<JsonRpcResponse> => {
        const reqId = ++nextId;
        const req: JsonRpcRequest = { method, id: reqId, params: rpcParams };
        proc!.stdin!.write(JSON.stringify(req) + "\n");

        return new Promise<JsonRpcResponse>((resolve, reject) => {
          const timer = setTimeout(() => {
            pending.delete(reqId);
            reject(new Error(`Codex request timeout: ${method}`));
          }, timeoutMs);
          pending.set(reqId, { resolve, reject, timer });
        });
      };

      const notify = (method: string, rpcParams?: Record<string, unknown>) => {
        const msg: JsonRpcNotification = { method, params: rpcParams };
        proc!.stdin!.write(JSON.stringify(msg) + "\n");
      };

      const respond = (requestId: string | number, result: unknown) => {
        proc!.stdin!.write(JSON.stringify({ id: requestId, result }) + "\n");
      };

      // ── Line processor ──────────────────────────────────────────────────

      const processLine = (line: string) => {
        if (!line.trim()) return;
        let msg: JsonRpcMessage;
        try {
          msg = JSON.parse(line);
        } catch {
          return;
        }

        // Response to a pending request
        if ("id" in msg && typeof msg.id === "number" && !("method" in msg)) {
          const pendingReq = pending.get(msg.id);
          if (pendingReq) {
            clearTimeout(pendingReq.timer);
            pending.delete(msg.id);
            if (msg.error) {
              pendingReq.reject(
                new Error(msg.error.message ?? "Codex RPC error"),
              );
            } else {
              pendingReq.resolve(msg as JsonRpcResponse);
            }
          }
          return;
        }

        // Server-to-client request. Approval policy is "never", but reject
        // defensively if a newer Codex still asks.
        if ("method" in msg && "id" in msg && msg.id !== undefined) {
          if (
            msg.method === "item/commandExecution/requestApproval" ||
            msg.method === "item/fileChange/requestApproval" ||
            msg.method === "execCommandApproval" ||
            msg.method === "applyPatchApproval"
          ) {
            respond(msg.id, { decision: "decline" });
          } else if (msg.method === "item/tool/requestUserInput") {
            respond(msg.id, { answers: {} });
          } else {
            proc!.stdin!.write(
              JSON.stringify({
                id: msg.id,
                error: { code: -32601, message: `Unsupported Codex server request: ${msg.method}` },
              }) + "\n",
            );
          }
          return;
        }

        // Notification
        const notification = msg as JsonRpcNotification;
        if (!notification.method) return;

        const p = notification.params ?? {};

        switch (notification.method) {
          case "thread/started": {
            threadId = (p as any)?.thread?.id;
            break;
          }

          case "turn/started": {
            turnId = (p as any)?.turn?.id;
            record.recentTools = [];
            break;
          }

          case "item/started": {
            const item = (p as any)?.item ?? p;
            if (isToolItem(item)) {
              const normalized = normalizeToolEvent({
                toolName:
                  item.name ?? item.command ?? item.type ?? "unknown",
                args: item.arguments ?? item.input ?? {},
                status: "start",
              });
              record.toolCount++;
              record.recentTools.push(normalized);
              if (record.recentTools.length > 20)
                record.recentTools.shift();
              onEvent({
                type: "subagent:progress",
                id,
                harness: "codex",
                timestamp: Date.now(),
                record,
              });
            }
            break;
          }

          case "item/completed": {
            const item = (p as any)?.item ?? p;
            if (item?.type === "agentMessage" && typeof item.text === "string") {
              outputText = item.text;
            }
            if (item?.type === "reasoning") {
              reasoningText = [...(item.summary ?? []), ...(item.content ?? [])].join("\n");
            }
            if (isToolItem(item)) {
              const match = findLastRunning(record.recentTools);
              if (match) match.status = "done";
              onEvent({
                type: "subagent:progress",
                id,
                harness: "codex",
                timestamp: Date.now(),
                record,
              });
            }
            break;
          }

          case "item/agentMessage/delta": {
            const delta = (p as any)?.delta;
            if (typeof delta === "string") outputText += delta;
            break;
          }

          case "item/reasoning/summaryTextDelta":
          case "item/reasoning/textDelta": {
            const delta = (p as any)?.delta;
            if (typeof delta === "string") reasoningText += delta;
            break;
          }

          case "thread/tokenUsage/updated": {
            // Notifications contain cumulative totals; assigning avoids double count.
            const usage = (p as any)?.tokenUsage?.total;
            if (usage) {
              record.usage.input = usage.inputTokens ?? 0;
              record.usage.output = usage.outputTokens ?? 0;
              record.usage.cacheRead = usage.cachedInputTokens ?? 0;
              record.usage.cacheWrite = 0;
            }
            break;
          }

          case "turn/completed": {
            const turn = (p as any)?.turn ?? p;
            const status = turn?.status ?? "completed";
            const finalMessage = Array.isArray(turn?.items)
              ? [...turn.items].reverse().find((item: any) => item?.type === "agentMessage")
              : undefined;
            if (typeof finalMessage?.text === "string") outputText = finalMessage.text;

            if (status === "completed") {
              turnResolve({ output: outputText, status });
            } else if (
              status === "interrupted" ||
              status === "failed" ||
              status === "cancelled"
            ) {
              turnReject(
                new Error(
                  `Turn ${status}${turn?.error ? ": " + turn.error : ""}`,
                ),
              );
            } else {
              // Unknown status, treat as completed
              turnResolve({ output: outputText, status });
            }
            break;
          }

          default:
            break;
        }
      };

      // ── Abort signal handling ───────────────────────────────────────────

      const abortPromise = new Promise<never>((_, reject) => {
        if (signal?.aborted) {
          reject(new Error("Aborted"));
          return;
        }
        abortHandler = () => {
          // Try graceful interrupt first
          if (turnId && threadId) {
            void request("turn/interrupt", { threadId, turnId }).catch(() => {});
          }
          reject(new Error("Aborted"));
        };
        signal?.addEventListener("abort", abortHandler, { once: true });
      });

      // ── Timeout promise ─────────────────────────────────────────────────

      const timeoutPromise = new Promise<never>((_, reject) => {
        turnTimer = setTimeout(() => {
          reject(new Error(`Codex turn timed out after ${TURN_TIMEOUT_MS}ms`));
        }, TURN_TIMEOUT_MS);
      });

      // ── Start reading lines (background) ────────────────────────────────
      // We read lines in the background but don't wait for readline close.
      let lineError: Error | null = null;
      let stderrText = "";
      proc.stderr?.on("data", (chunk: Buffer) => {
        stderrText = (stderrText + chunk.toString()).slice(-8_000);
      });
      rl.on("line", (line: string) => {
        try {
          processLine(line);
        } catch {
          // Line processing errors shouldn't kill the harness
        }
      });
      rl.on("close", () => {
        const closeError =
          lineError ??
          new Error(
            `Codex app-server closed before turn completion${stderrText.trim() ? `: ${stderrText.trim()}` : ""}`,
          );
        for (const request of pending.values()) {
          clearTimeout(request.timer);
          request.reject(closeError);
        }
        pending.clear();
        if (!turnSettled) turnReject(closeError);
      });
      proc.on("error", (err) => {
        lineError = err;
        turnReject(err);
      });

      // ── Protocol handshake ──────────────────────────────────────────────

      // 1. initialize
      await request("initialize", {
        clientInfo: {
          name: "pi_subagents",
          title: "Pi Subagent Manager",
          version: "0.2.0",
        },
        capabilities: {
          experimentalApi: true,
          requestAttestation: false,
          optOutNotificationMethods: [
            "thread/archived",
            "thread/unarchived",
            "thread/status/changed",
          ],
        },
      });

      // 2. initialized notification
      notify("initialized");

      // 3. thread/start
      const threadResp = await request("thread/start", {
        ...(model ? { model } : {}),
        cwd,
        approvalPolicy: "never",
        sandbox,
        ephemeral: true,
      });
      threadId =
        (threadResp.result as any)?.thread?.id ?? threadId;

      // 4. turn/start
      await request("turn/start", {
        threadId: threadId!,
        input: [
          {
            type: "text",
            text: params.task,
            text_elements: [] as string[],
          },
        ],
        effort,
      });

      // ── Wait for turn completion (race turn vs abort vs timeout) ─────────

      let turnResult: { output: string; status: string };
      try {
        turnResult = await Promise.race([
          turnPromise,
          abortPromise,
          timeoutPromise,
        ]);
      } catch (err) {
        // Turn failed, interrupted, timed out, or aborted
        const errMsg = err instanceof Error ? err.message : String(err);
        record.completedAt = Date.now();

        if (signal?.aborted) {
          record.status = "cancelled";
          record.error = "Cancelled";
        } else {
          record.status = "failed";
          record.error = errMsg;
        }

        // Build best-effort output
        if (reasoningText) {
          record.output =
            "[Reasoning]\n" + reasoningText +
            "\n\n[Output]\n" + outputText;
        } else {
          record.output = outputText || "(no output)";
        }

        onEvent({
          type:
            record.status === "cancelled"
              ? "subagent:cancelled"
              : "subagent:failed",
          id,
          harness: "codex",
          timestamp: record.completedAt,
          record,
          error: record.error,
        });
        return;
      } finally {
        if (turnTimer) clearTimeout(turnTimer);
        if (abortHandler) signal?.removeEventListener("abort", abortHandler);
      }

      // ── Turn completed successfully ──────────────────────────────────────

      outputText = turnResult.output || outputText;
      record.usage.turns = 1;

      if (reasoningText) {
        record.output =
          "[Reasoning]\n" + reasoningText +
          "\n\n[Output]\n" + outputText;
      } else {
        record.output = outputText || "(no output)";
      }
      record.completedAt = Date.now();

      if (signal?.aborted) {
        record.status = "cancelled";
        onEvent({
          type: "subagent:cancelled",
          id,
          harness: "codex",
          timestamp: record.completedAt,
          record,
        });
      } else {
        record.status = "completed";
        onEvent({
          type: "subagent:completed",
          id,
          harness: "codex",
          timestamp: record.completedAt,
          record,
        });
      }
    } catch (err) {
      // Top-level error (e.g., spawn failure, RPC failure during setup)
      record.completedAt = Date.now();
      if (signal?.aborted) {
        record.status = "cancelled";
        onEvent({
          type: "subagent:cancelled",
          id,
          harness: "codex",
          timestamp: record.completedAt,
          record,
        });
      } else {
        record.status = "failed";
        record.error = err instanceof Error ? err.message : String(err);
        onEvent({
          type: "subagent:failed",
          id,
          harness: "codex",
          timestamp: record.completedAt,
          record,
          error: record.error,
        });
      }
    } finally {
      if (turnTimer) clearTimeout(turnTimer);
      if (abortHandler) signal?.removeEventListener("abort", abortHandler);
      // Always clean up the process
      if (proc) {
        this.activeProcesses.delete(id);
        await terminateProcessTree(proc);
      }
      if (codexSqliteHome) {
        await rm(codexSqliteHome, {
          recursive: true,
          force: true,
          maxRetries: 10,
          retryDelay: 200,
        }).catch(() => {});
      }
    }
  }

  async cancel(id: string): Promise<void> {
    const proc = this.activeProcesses.get(id);
    if (proc) {
      await terminateProcessTree(proc);
      this.activeProcesses.delete(id);
    }
  }

  dispose(): void {
    for (const [id, proc] of this.activeProcesses) {
      terminateProcessTree(proc).catch(() => {});
    }
    this.activeProcesses.clear();
  }
}

// ── Helpers ──────────────────────────────────────────────────────────────────

function clampEffort(
  requested: string,
  max: "low" | "medium" | "high",
): "low" | "medium" | "high" {
  const levels: Array<"low" | "medium" | "high"> = [
    "low",
    "medium",
    "high",
  ];
  const reqIdx = levels.indexOf(requested as any);
  const maxIdx = levels.indexOf(max);
  if (reqIdx === -1) return max;
  return levels[Math.min(reqIdx, maxIdx)];
}

function isToolItem(item: any): boolean {
  return [
    "commandExecution",
    "fileChange",
    "mcpToolCall",
    "dynamicToolCall",
    "collabAgentToolCall",
  ].includes(item?.type);
}

function findLastRunning(
  tools: Array<{ status: "running" | "done" }>,
): { status: "running" | "done" } | undefined {
  for (let i = tools.length - 1; i >= 0; i--) {
    if (tools[i].status === "running") return tools[i];
  }
  return undefined;
}