// `machine <name>`: use a machine directly, in Pi's interactive terminal UI.
//
// Pi shows the conversation; the machine supplies everything else. The model
// sees exactly the machine's instructions and tools (as under the pi harness),
// tools run in this process against a sandbox over the current directory, and
// every exchange (a prompt and everything the agent did for it) is stored as a
// run with its full trajectory, so interactive use produces the same records
// that suites, grading, and training read. Slash commands reach back into the
// machine:
//
//   /score            grade the last exchange against the machine's rubric
//   /good, /bad       mark the last exchange (stored as feedback on its run)
//   /correct <note>   record what should change ("too terse"), for later rubric
//                     and case proposals
//   /machine          show the machine, its model, and this session's runs and spend
//
// Pi's own built-in tools, extensions, skills, context files, and ~/.pi setup
// stay out, as in the harness. Pi's state for these sessions lives under
// .machine/pi/, so `--continue` resumes the last conversation with a machine.

import { spawn } from "node:child_process";
import { randomBytes } from "node:crypto";
import { mkdir, writeFile } from "node:fs/promises";
import { createServer, type IncomingMessage, type Server } from "node:http";
import path from "node:path";
import { piEnvironment, piResult, resolvePiModel, TINKER_PROVIDER, type PiMessage } from "./harnesses/pi.ts";
import { TINKER_OAI_BASE_URL } from "./harnesses/chat-completions.ts";
import type { Machine } from "./machine.ts";
import type { Tool, ToolContext } from "./harness.ts";
import type { Price } from "./pricing.ts";
import { costOf } from "./pricing.ts";
import { directorySandbox, type Isolation } from "./sandbox.ts";
import { preview } from "./events.ts";
import { nameSegments, newRunId } from "./store.ts";
import type { Feedback, Message, Run } from "./types.ts";
import type { Workspace } from "./workspace.ts";

export type InteractiveOptions = {
  /** Where the session works: tools act here. Defaults to the current directory. */
  cwd?: string;
  /** Model override; defaults to the machine's. */
  model?: string;
  /** The machine version recorded on each exchange's run, e.g. "v2". */
  version?: string;
  /** Resume the last Pi session with this machine. */
  continue?: boolean;
  /** Grade every exchange as it finishes (judge graders cost money). */
  evaluate?: boolean;
  /** Price of a model, for the spend shown by /machine. */
  price?: (model: string) => Price | undefined;
  isolation?: Isolation;
  /** The Pi executable. */
  command?: string;
  /** Directory for Pi's state for machine sessions (agent dir and session files). */
  stateDir: string;
  /** Extra arguments for pi, e.g. ["-p", "hello"] to run one prompt non-interactively (tests, scripts). */
  piArgs?: string[];
};

type Exchange = { run: Run };

/** Open the machine in Pi and block until the user quits. Returns Pi's exit code. */
export async function interactive(machine: Machine, workspace: Workspace, options: InteractiveOptions): Promise<number> {
  const cwd = path.resolve(options.cwd ?? process.cwd());
  const model = options.model ?? machine.model;
  const target = resolvePiModel(model);
  if (target.tinker && !process.env.TINKER_API_KEY) throw new Error(`${model} runs on Tinker; set TINKER_API_KEY.`);

  const sandbox = await directorySandbox(cwd, { isolation: options.isolation });
  const session = `${path.basename(cwd).toLowerCase().replace(/[^a-z0-9._-]+/g, "-") || "session"}-${newRunId()}`;
  const exchanges: Exchange[] = [];
  let history: Message[] = [];
  let prompt: string | undefined;

  const record = async (messages: PiMessage[]): Promise<Run | undefined> => {
    const input = userText(messages);
    if (input === undefined) return undefined;
    let result;
    try {
      result = piResult(input, messages.filter((m) => m.role !== "user"));
    } catch {
      return undefined; // a failed attempt Pi will retry or report itself
    }
    const usage = result.usage && !result.usage.costUsd && options.price?.(model) ? { ...result.usage, costUsd: costOf(options.price(model)!, result.usage) } : result.usage;
    const now = new Date().toISOString();
    let run: Run = {
      id: newRunId(),
      machine: machine.name,
      ...(options.version ? { version: options.version } : {}),
      workspace: workspace.name,
      instructions: machine.instructions,
      input,
      history,
      trajectory: result.trajectory,
      output: result.output,
      model,
      harness: "pi-interactive",
      usage,
      stopReason: result.stopReason,
      startedAt: now,
      finishedAt: now,
      session,
    };
    await workspace.store.putRun(run);
    workspace.events.emit({ type: "run_end", machine: machine.name, run: run.id, model, session, stopReason: run.stopReason, costUsd: usage?.costUsd, input: preview(input), output: preview(run.output) });
    history = [...history, ...result.trajectory];
    prompt = undefined;
    await workspace.store.putSession({ machine: machine.name, name: session, messages: history, updatedAt: now });
    if (options.evaluate) run = await workspace.evaluate(machine, run);
    exchanges.push({ run });
    return run;
  };

  const last = () => exchanges.at(-1);
  const addFeedback = async (feedback: Feedback) => {
    const exchange = last();
    if (!exchange) return "Nothing to mark yet: no exchange has finished in this session.";
    exchange.run = { ...exchange.run, feedback: [...(exchange.run.feedback ?? []), feedback] };
    await workspace.store.putRun(exchange.run);
    workspace.events.emit({ type: "feedback", machine: machine.name, run: exchange.run.id, kind: feedback.kind, note: feedback.note });
    return `Marked ${exchange.run.id} ${feedback.kind}${feedback.note ? `: ${feedback.note}` : ""}.`;
  };

  // Tools run here, in the machine process, against the session's sandbox. Calls stream
  // progress back to Pi (and keep the connection alive) while a long tool runs.
  const byName = new Map(machine.tools.map((t) => [t.name, t]));
  const runTool = (body: { name?: string; arguments?: unknown }, send: (line: ToolStreamLine) => void, signal: AbortSignal) => {
    const tool = byName.get(body.name ?? "");
    if (!tool) return Promise.resolve(send({ result: `Unknown tool "${body.name}".`, isError: true }));
    const conversation = () => [...history, ...(prompt === undefined ? [] : [{ role: "user" as const, content: prompt }])];
    return streamToolCall(tool, body.arguments ?? {}, { sandbox, interactive: true, conversation, signal }, send, {
      onStart: (input) => workspace.events.emit({ type: "tool_call", machine: machine.name, session, name: tool.name, input }),
      onEnd: (text) => workspace.events.emit({ type: "tool_result", machine: machine.name, session, name: tool.name, preview: preview(text) }),
    });
  };

  const control = await serveControl(runTool, async (action, body) => {
    switch (action) {
      case "start":
        prompt = body.note ?? "";
        workspace.events.emit({ type: "run_start", machine: machine.name, model, harness: "pi-interactive", session, input: preview(body.note ?? "") });
        return "";
      case "record": {
        const run = await record(body.messages ?? []);
        return run?.scorecard ? `${scoreLine(run)}` : "";
      }
      case "score": {
        const exchange = last();
        if (!exchange) return "Nothing to score yet: no exchange has finished in this session.";
        exchange.run = await workspace.evaluate(machine, exchange.run);
        return scorecardText(exchange.run);
      }
      case "good":
      case "bad":
        return addFeedback({ kind: action, note: body.note || undefined, at: new Date().toISOString() });
      case "correct":
        if (!body.note?.trim()) return "Usage: /correct <what should change>, e.g. /correct too terse";
        return addFeedback({ kind: "correction", note: body.note.trim(), at: new Date().toISOString() });
      case "machine": {
        const spent = exchanges.reduce((s, e) => s + (e.run.usage?.costUsd ?? 0) + (e.run.scorecard?.usage?.costUsd ?? 0), 0);
        return [
          `${machine.name}${options.version ? ` ${options.version}` : ""} · ${model}`,
          machine.intent,
          `criteria: ${Object.keys(machine.rubric).join(", ")}`,
          `tools: ${machine.tools.map((t) => t.name).join(", ") || "none"}`,
          `workspace: ${cwd}${sandbox.isolated ? " (commands isolated)" : " (commands NOT isolated: bubblewrap unavailable)"}`,
          `session ${session}: ${exchanges.length} exchange${exchanges.length === 1 ? "" : "s"} stored, about $${spent.toFixed(4)}`,
        ].join("\n");
      }
      default:
        throw new Error(`unknown action ${action}`);
    }
  });

  try {
    const state = path.join(options.stateDir, ...nameSegments(machine.name, "machine name"));
    const agentDir = path.join(state, "agent");
    await mkdir(agentDir, { recursive: true, mode: 0o700 });
    const extensionFile = path.join(state, "machine-session.js");
    const systemPromptFile = path.join(state, "system-prompt.md");
    await writeFile(systemPromptFile, machine.instructions, { mode: 0o600 });
    await writeFile(
      extensionFile,
      interactiveExtensionSource({
        systemPrompt: machine.instructions,
        tools: machine.tools.map((t) => ({ name: t.name, description: t.description, inputSchema: t.inputSchema })),
        provider: target.tinker ? tinkerProviderConfig(target.model) : null,
      }),
      { mode: 0o600 },
    );
    const env = {
      ...piEnvironment(agentDir, undefined, process.env, { attribution: target.provider === "openrouter" }),
      MACHINE_PI_CONTROL_URL: control.url,
      MACHINE_PI_CONTROL_TOKEN: control.token,
    };
    const args = [
      "--offline",
      "--no-extensions",
      "--extension", extensionFile,
      "--no-skills",
      "--no-prompt-templates",
      "--no-themes",
      "--no-context-files",
      "--no-approve",
      "--session-dir", path.join(state, "sessions"),
      "--system-prompt", systemPromptFile,
      ...(machine.tools.length > 0 ? ["--tools", machine.tools.map((t) => t.name).join(",")] : ["--no-tools"]),
      "--provider", target.provider,
      "--model", target.model,
      ...(options.continue ? ["--continue"] : []),
      ...(options.piArgs ?? []),
    ];
    return await new Promise<number>((resolve, reject) => {
      const proc = spawn(options.command ?? "pi", args, { cwd, env, stdio: "inherit" });
      proc.once("error", (error) => reject(new Error(`Could not start pi (is Pi installed and on PATH?): ${error.message}`)));
      proc.once("exit", (code, signal) => resolve(code ?? (signal ? 1 : 0)));
    });
  } finally {
    await control.close();
    if (exchanges.length) console.error(`${machine.name}: ${exchanges.length} exchange${exchanges.length === 1 ? "" : "s"} stored in session ${session}.`);
  }
}

function userText(messages: PiMessage[]): string | undefined {
  const user = messages.find((m) => m.role === "user");
  if (!user) return undefined;
  if (typeof user.content === "string") return user.content;
  return (user.content ?? []).map((c) => (c.type === "text" && typeof (c as { text?: unknown }).text === "string" ? (c as { text: string }).text : "")).join("");
}

function scoreLine(run: Run): string {
  const failing = Object.entries(run.scorecard!.criteria).filter(([, c]) => c.score < 1).map(([id]) => id);
  return `score ${run.scorecard!.total.toFixed(2)}${failing.length ? ` (below 1: ${failing.join(", ")})` : ""}`;
}

function scorecardText(run: Run): string {
  const sc = run.scorecard!;
  return [`score ${sc.total.toFixed(2)}`, ...Object.entries(sc.criteria).map(([id, c]) => `${id}: ${c.score.toFixed(2)} (${c.grader}) ${c.reason}`)].join("\n");
}

/** One line of a streamed tool call: progress while it runs, a keep-alive tick, then the result. */
export type ToolStreamLine = { progress: string } | { tick: true } | { result: string; isError?: boolean };

/**
 * Run a tool and report it as stream lines: each progress line the tool
 * reports, a tick every `tickMs` so the connection never sits idle (HTTP
 * clients time out an idle response), then the result or the error.
 */
export async function streamToolCall(
  tool: Tool,
  input: unknown,
  context: ToolContext,
  send: (line: ToolStreamLine) => void,
  hooks: { onStart?: (input: unknown) => void; onEnd?: (text: string) => void; tickMs?: number } = {},
): Promise<void> {
  hooks.onStart?.(input);
  const tick = setInterval(() => send({ tick: true }), hooks.tickMs ?? 15_000);
  try {
    const result = await tool.execute(input, { ...context, progress: (line) => send({ progress: line }) });
    const text = typeof result === "string" ? result : JSON.stringify(result);
    hooks.onEnd?.(text);
    send({ result: text });
  } catch (error) {
    const text = error instanceof Error ? error.message : String(error);
    hooks.onEnd?.(text);
    send({ result: text, isError: true });
  } finally {
    clearInterval(tick);
  }
}

/**
 * A loopback endpoint the session's Pi extension calls: POST /<action> with a
 * bearer token, JSON in, text out; POST /tool streams a tool call as JSON lines.
 */
async function serveControl(
  runTool: (body: { name?: string; arguments?: unknown }, send: (line: ToolStreamLine) => void, signal: AbortSignal) => Promise<void>,
  handle: (action: string, body: { messages?: PiMessage[]; note?: string }) => Promise<string>,
) {
  const token = randomBytes(24).toString("hex");
  const server: Server = createServer(async (req, res) => {
    if (req.method !== "POST" || req.headers.authorization !== `Bearer ${token}`) {
      res.writeHead(req.method === "POST" ? 401 : 405).end();
      return;
    }
    try {
      const body = JSON.parse((await readBody(req)) || "{}");
      const action = (req.url ?? "/").slice(1);
      if (action === "tool") {
        const abort = new AbortController();
        res.on("close", () => abort.abort());
        res.writeHead(200, { "content-type": "application/x-ndjson; charset=utf-8" });
        await runTool(body, (line) => void (res.writableEnded || res.write(`${JSON.stringify(line)}\n`)), abort.signal);
        res.end();
        return;
      }
      const text = await handle(action, body);
      res.writeHead(200, { "content-type": "text/plain; charset=utf-8" }).end(text);
    } catch (error) {
      if (res.headersSent) res.end();
      else res.writeHead(500, { "content-type": "text/plain; charset=utf-8" }).end(error instanceof Error ? error.message : String(error));
    }
  });
  await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
  const address = server.address();
  const port = typeof address === "object" && address ? address.port : 0;
  return {
    url: `http://127.0.0.1:${port}`,
    token,
    close: () =>
      new Promise<void>((resolve) => {
        server.closeAllConnections?.();
        server.close(() => resolve());
      }),
  };
}

function readBody(req: IncomingMessage): Promise<string> {
  return new Promise((resolve, reject) => {
    let data = "";
    req.setEncoding("utf8");
    req.on("data", (chunk) => (data += chunk));
    req.on("end", () => resolve(data));
    req.on("error", reject);
  });
}

function tinkerProviderConfig(model: string) {
  return {
    name: TINKER_PROVIDER,
    config: {
      name: "Tinker",
      baseUrl: TINKER_OAI_BASE_URL,
      apiKey: "$TINKER_API_KEY",
      api: "openai-completions",
      models: [
        {
          id: model,
          name: model,
          reasoning: false,
          input: ["text"],
          cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
          contextWindow: 65_536,
          maxTokens: 8_192,
          compat: { supportsStore: false, supportsDeveloperRole: false, supportsReasoningEffort: false, supportsStrictMode: false, maxTokensField: "max_tokens" },
        },
      ],
    },
  };
}

/** Source of the session's Pi extension: the machine's prompt, tools, provider, record hook, and slash commands. */
export function interactiveExtensionSource(spec: {
  systemPrompt: string;
  tools: Array<{ name: string; description: string; inputSchema: Record<string, unknown> }>;
  provider: { name: string; config: Record<string, unknown> } | null;
}): string {
  return `// Generated by machine for an interactive session; rewritten each time the machine is opened.
const SPEC = ${JSON.stringify(spec)};

export default function (pi) {
  const controlUrl = process.env.MACHINE_PI_CONTROL_URL;
  const controlToken = process.env.MACHINE_PI_CONTROL_TOKEN;
  if (SPEC.provider) pi.registerProvider(SPEC.provider.name, SPEC.provider.config);

  // The machine's instructions are the whole system prompt.
  pi.on("before_agent_start", (event) => {
    control("start", { note: event?.prompt ?? "" }).catch(() => {});
    return { systemPrompt: SPEC.systemPrompt };
  });

  async function control(action, body) {
    const response = await fetch(controlUrl + "/" + action, {
      method: "POST",
      headers: { "content-type": "application/json", authorization: "Bearer " + controlToken },
      body: JSON.stringify(body ?? {}),
    });
    const text = await response.text();
    if (!response.ok) throw new Error(text || "machine returned HTTP " + response.status);
    return text;
  }

  // Store each finished exchange as a run.
  pi.on("agent_end", async (event, ctx) => {
    try {
      const note = await control("record", { messages: event.messages });
      if (note) ctx.ui.notify(note, "info");
    } catch (error) {
      ctx.ui.notify("machine could not store this exchange: " + (error instanceof Error ? error.message : String(error)), "error");
    }
  });

  const commands = {
    score: "Grade the last exchange against the machine's rubric",
    good: "Mark the last exchange good",
    bad: "Mark the last exchange bad",
    correct: "Record what should change about the last exchange: /correct <note>",
    machine: "Show the machine, its model, and this session",
  };
  for (const [name, description] of Object.entries(commands)) {
    pi.registerCommand(name, {
      description,
      handler: async (args, ctx) => {
        try {
          if (name === "score") ctx.ui.notify("grading…", "info");
          ctx.ui.notify(await control(name, { note: (args ?? "").trim() }), "info");
        } catch (error) {
          ctx.ui.notify(error instanceof Error ? error.message : String(error), "error");
        }
      },
    });
  }

  // A tool call streams JSON lines: progress (shown in Pi as it happens), ticks, then the result.
  async function callTool(name, args, signal, onUpdate) {
    try {
      const response = await fetch(controlUrl + "/tool", {
        method: "POST",
        headers: { "content-type": "application/json", authorization: "Bearer " + controlToken },
        body: JSON.stringify({ name, arguments: args ?? {} }),
        signal,
      });
      if (!response.ok || !response.body) return { text: JSON.stringify({ error: "machine tool endpoint returned HTTP " + response.status }) };
      const reader = response.body.getReader();
      const decoder = new TextDecoder();
      const progress = [];
      let buffer = "";
      for (;;) {
        const { value, done } = await reader.read();
        if (value) buffer += decoder.decode(value, { stream: true });
        let newline;
        while ((newline = buffer.indexOf("\\n")) >= 0) {
          const line = buffer.slice(0, newline).trim();
          buffer = buffer.slice(newline + 1);
          if (!line) continue;
          const message = JSON.parse(line);
          if (typeof message.progress === "string") {
            progress.push(message.progress);
            if (onUpdate) onUpdate({ content: [{ type: "text", text: progress.slice(-12).join("\\n") }], details: {} });
          } else if (typeof message.result === "string") {
            return { text: message.isError ? JSON.stringify({ error: message.result }) : message.result };
          }
        }
        if (done) return { text: JSON.stringify({ error: "the machine ended the tool call without a result" }) };
      }
    } catch (error) {
      return { text: JSON.stringify({ error: error instanceof Error ? error.message : String(error) }) };
    }
  }

  let queue = Promise.resolve();
  for (const tool of SPEC.tools) {
    pi.registerTool({
      name: tool.name,
      label: tool.name,
      description: tool.description,
      parameters: tool.inputSchema,
      execute(_toolCallId, params, signal, onUpdate) {
        const result = queue.then(() => callTool(tool.name, params, signal, onUpdate));
        queue = result.catch(() => {});
        return result.then(({ text }) => ({ content: [{ type: "text", text }], details: {} }));
      },
    });
  }
}
`;
}