import { spawn } from "node:child_process";
import { createHash } from "node:crypto";
import { statSync } from "node:fs";
import {
  appendFile,
  chmod,
  copyFile,
  cp,
  mkdir,
  readdir,
  readFile,
  realpath,
  rm,
  stat,
  writeFile,
} from "node:fs/promises";
import { join } from "node:path";
import { query, type SDKMessage } from "@anthropic-ai/claude-agent-sdk";
import type { ArtifactKind, RunEventInput, RunManifest, StageKey } from "@autoscholar/schemas";
import { MetricsFile } from "@autoscholar/schemas";
import { MAX_ARTIFACT_BYTES, syncArtifacts, writeChecksums } from "./artifacts.js";
import { buildLatex, listFiles, renderPages } from "./latex.js";
import {
  createPolicyHook,
  PLATFORM_SYSTEM_APPEND,
  RESEARCH_ALLOWED_TOOLS,
  RESEARCH_DISALLOWED_TOOLS,
  sandboxSettings,
} from "./policy.js";
import { STAGE_SPECS, type StageContext, type StageSpec } from "./stages.js";
import { createResearchTools } from "./tools.js";

export interface RunnerJob {
  runId: string;
  attemptNo: number;
  manifest: RunManifest;
  instructionText: string;
  resumeFromStage: StageKey | null;
  checkpoint: Record<string, unknown> | null;
}

export interface RunnerHost {
  workspaceRoot: string;
  apiKey: string;
  extraDomains: string[];
  tectonicBin?: string;
  tectonicCache?: string;
  latexTemplateDir: string;
  workerName: string;
  sdkVersion: string;
  toolchain: Record<string, string>;
  emit: (ev: RunEventInput) => void;
  stage: (report: {
    stage: StageKey;
    status: "running" | "completed" | "failed" | "skipped";
    costUsd: number;
    turns: number;
    usage?: Record<string, unknown>;
    error?: string;
    checkpoint?: Record<string, unknown>;
  }) => Promise<void>;
  upload: (path: string, kind: ArtifactKind, contentType: string, buf: Buffer) => Promise<void>;
  isCancelled: () => boolean;
  /** Optional: fetch the latest artifacts of this run from the control plane (resume on a different node). */
  fetchArtifacts?: () => Promise<Array<{ path: string; read: () => Promise<Buffer> }>>;
}

export interface RunOutcome {
  outcome: "completed" | "failed" | "cancelled" | "budget_exceeded" | "timeout";
  error?: string;
  costUsd: number;
  turns: number;
  stagesCompleted: StageKey[];
  resourceUsage: { wallSeconds: number; diskBytes: number; peakRssBytes: number };
}

interface SessionResult {
  costUsd: number;
  turns: number;
  subtype: string;
  resultText: string;
  usage: Record<string, unknown>;
  denied: number;
}

async function dirSize(dir: string): Promise<number> {
  let total = 0;
  for (const f of await listFiles(dir)) {
    try {
      total += (await stat(join(dir, f))).size;
    } catch {
      /* ignore */
    }
  }
  return total;
}

export class ResearchRunner {
  private readonly ws: string;
  private spent = 0;
  private turns = 0;
  private readonly started = Date.now();
  private completed: StageKey[] = [];
  private notes: string[] = [];
  private peakRss = 0;

  constructor(
    private readonly job: RunnerJob,
    private readonly host: RunnerHost,
  ) {
    this.ws = join(host.workspaceRoot, "runs", job.runId);
  }

  private emit(
    kind: string,
    message: string,
    data?: Record<string, unknown>,
    level: RunEventInput["level"] = "info",
    stage?: StageKey,
  ) {
    const ev: RunEventInput = {
      kind,
      message: message.slice(0, 4000),
      data,
      level,
      stage,
      at: new Date().toISOString(),
    };
    this.host.emit(ev);
    appendFile(
      join(this.ws, "provenance/events.jsonl"),
      `${JSON.stringify({ ...ev, runId: this.job.runId, attemptNo: this.job.attemptNo })}\n`,
    ).catch(() => {});
    const rss = process.memoryUsage().rss;
    if (rss > this.peakRss) this.peakRss = rss;
  }

  private remainingBudget(): number {
    return this.job.manifest.maxBudgetUsd - this.spent;
  }
  private remainingMinutes(): number {
    return this.job.manifest.maxRuntimeMinutes - (Date.now() - this.started) / 60_000;
  }

  /** Creates the isolated workspace: frozen CLAUDE.md as project instructions, contract directories, LaTeX class. */
  async prepareWorkspace(): Promise<{ resumed: boolean; startIndex: number }> {
    const ws = this.ws;
    let resumed = false;
    let startIndex = 0;
    const existed = await stat(ws)
      .then((s) => s.isDirectory())
      .catch(() => false);
    const cpStage = (this.job.checkpoint?.stage as StageKey | undefined) ?? this.job.resumeFromStage;
    if (existed && cpStage && STAGE_SPECS.some((s) => s.key === cpStage)) {
      const idx = STAGE_SPECS.findIndex((s) => s.key === cpStage);
      // verify the frozen instruction is intact; otherwise restart from scratch
      const frozen = await readFile(join(ws, "CLAUDE.md"), "utf8").catch(() => "");
      if (createHash("sha256").update(frozen).digest("hex") === this.job.manifest.instructionSha256) {
        resumed = true;
        startIndex = idx + 1;
        this.completed =
          (this.job.checkpoint?.completedStages as StageKey[] | undefined) ??
          STAGE_SPECS.slice(0, idx + 1).map((s) => s.key);
        this.spent = Number(this.job.checkpoint?.spentUsd ?? 0);
      }
    }
    if (!resumed) {
      if (existed) {
        await chmod(join(ws, "input"), 0o755).catch(() => {});
        for (const f of ["CLAUDE.md", "input/CLAUDE.md", "input/run.json"])
          await chmod(join(ws, f), 0o644).catch(() => {});
        await rm(ws, { recursive: true, force: true });
      }
      for (const d of [
        "input",
        "plan",
        "sources",
        "data",
        "code",
        "results/tables",
        "results/figures",
        "manuscript",
        "review",
        "provenance",
        "home/.claude",
        ".autoscholar",
        ".cache/uv",
        ".cache/tmp",
      ])
        await mkdir(join(ws, d), { recursive: true });
      await writeFile(join(ws, "CLAUDE.md"), this.job.instructionText);
      await writeFile(join(ws, "input/CLAUDE.md"), this.job.instructionText);
      await chmod(join(ws, "CLAUDE.md"), 0o444);
      await chmod(join(ws, "input/CLAUDE.md"), 0o444);
      await writeFile(
        join(ws, "input/run.json"),
        `${JSON.stringify({ ...this.job.manifest, workerName: this.host.workerName, attemptNo: this.job.attemptNo, frozenAt: new Date().toISOString() }, null, 2)}\n`,
      );
      await chmod(join(ws, "input/run.json"), 0o444);
      await chmod(join(ws, "input"), 0o555);
      await writeFile(
        join(ws, "home/.claude/settings.json"),
        JSON.stringify({ env: {}, includeCoAuthoredBy: false }, null, 2),
      );
      for (const f of await readdir(this.host.latexTemplateDir).catch(() => [] as string[])) {
        if (/\.(cls|sty|bst|mplstyle)$/.test(f))
          await copyFile(join(this.host.latexTemplateDir, f), join(ws, "manuscript", f));
      }
      await writeFile(join(ws, ".gitignore"), "home/\n.cache/\n.autoscholar/\n");
    }
    this.emit(
      "workspace.ready",
      resumed ? `resumed workspace at stage index ${startIndex}` : "fresh isolated workspace prepared",
      { workspace: ws, resumed, instructionSha256: this.job.manifest.instructionSha256 },
    );
    return { resumed, startIndex };
  }

  private buildOptions(
    spec: StageSpec,
    maxBudgetUsd: number,
    maxTurns: number,
    abort: AbortController,
    denied: { n: number },
  ) {
    const ws = this.ws;
    const reviewer = !!spec.reviewer;
    const hook = createPolicyHook({
      workspace: ws,
      frozenPaths: ["CLAUDE.md", "input", "provenance", ".gitignore"],
      writableOnly: reviewer ? ["review/"] : undefined,
      onDenied: (tool, reason, input) => {
        denied.n++;
        this.emit("policy.denied", `${tool} denied: ${reason}`, { tool, reason, input }, "warn", spec.key);
      },
    });
    const tools = createResearchTools({
      workspace: ws,
      tectonicBin: this.host.tectonicBin,
      tectonicCache: this.host.tectonicCache,
      emit: (k, m, d) => this.emit(k, m, d, "info", spec.key),
    });
    const reviewerAppend = reviewer
      ? "\n\n# Reviewer session\nYou are the independent referee for this manuscript. You did not author it. You may only write files under review/. Be rigorous, specific and fair; blockers only for fabricated/unverifiable/invalid science."
      : "";
    return {
      cwd: ws,
      model: reviewer ? this.job.manifest.reviewerModel : this.job.manifest.model,
      systemPrompt: {
        type: "preset" as const,
        preset: "claude_code" as const,
        append: PLATFORM_SYSTEM_APPEND + reviewerAppend,
      },
      settingSources: ["project" as const],
      permissionMode: "dontAsk" as const,
      allowedTools: RESEARCH_ALLOWED_TOOLS,
      disallowedTools: RESEARCH_DISALLOWED_TOOLS,
      hooks: { PreToolUse: [{ hooks: [hook] }] },
      sandbox: sandboxSettings(ws, this.host.extraDomains),
      mcpServers: { autoscholar: tools },
      maxBudgetUsd,
      maxTurns,
      abortController: abort,
      persistSession: false,
      includePartialMessages: false,
      env: {
        PATH: "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin",
        HOME: join(ws, "home"),
        CLAUDE_CONFIG_DIR: join(ws, "home/.claude"),
        ANTHROPIC_API_KEY: this.host.apiKey,
        TMPDIR: join(ws, ".cache/tmp"),
        UV_CACHE_DIR: join(ws, ".cache/uv"),
        TECTONIC_CACHE_DIR: this.host.tectonicCache ?? join(ws, ".cache/tectonic"),
        LANG: "en_US.UTF-8",
        LC_ALL: "en_US.UTF-8",
        SOURCE_DATE_EPOCH: "1700000000",
        MPLBACKEND: "Agg",
        PYTHONDONTWRITEBYTECODE: "1",
        CLAUDE_CODE_DISABLE_AUTO_MEMORY: "1",
        DISABLE_TELEMETRY: "1",
        DISABLE_AUTOUPDATER: "1",
        CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC: "1",
        AUTOSCHOLAR_RUN_ID: this.job.runId,
      },
      stderr: (d: string) => {
        if (/error|fatal|sandbox/i.test(d))
          this.emit("sdk.stderr", d.slice(0, 500), undefined, "debug", spec.key);
      },
    };
  }

  private async session(
    spec: StageSpec,
    prompt: string,
    limits: { maxBudgetUsd: number; maxTurns: number; timeoutMs: number },
  ): Promise<SessionResult> {
    const abort = new AbortController();
    const denied = { n: 0 };
    const timer = setTimeout(() => abort.abort(new Error("stage wall-clock limit")), limits.timeoutMs);
    const cancelPoll = setInterval(() => {
      if (this.host.isCancelled()) abort.abort(new Error("cancelled"));
    }, 5000);
    const result: SessionResult = {
      costUsd: 0,
      turns: 0,
      subtype: "unknown",
      resultText: "",
      usage: {},
      denied: 0,
    };
    const t0 = Date.now();
    try {
      const q = query({
        prompt,
        options: this.buildOptions(spec, limits.maxBudgetUsd, limits.maxTurns, abort, denied),
      });
      let toolCalls = 0;
      for await (const msg of q as AsyncIterable<SDKMessage>) {
        if (msg.type === "system" && (msg as { subtype?: string }).subtype === "init") {
          const m = msg as unknown as {
            model?: string;
            claude_code_version?: string;
            tools?: string[];
            mcp_servers?: Array<{ name: string; status: string }>;
          };
          this.emit(
            "session.init",
            `session started (${m.model ?? "model"}; claude code ${m.claude_code_version ?? "?"})`,
            { tools: m.tools?.length, mcp: m.mcp_servers },
            "debug",
            spec.key,
          );
        } else if (msg.type === "assistant") {
          const content =
            (
              msg as unknown as {
                message: {
                  content: Array<{
                    type: string;
                    text?: string;
                    name?: string;
                    input?: Record<string, unknown>;
                  }>;
                };
              }
            ).message.content ?? [];
          for (const c of content) {
            if (c.type === "text" && c.text?.trim())
              this.emit("assistant.text", c.text.slice(0, 600), undefined, "debug", spec.key);
            if (c.type === "tool_use") {
              toolCalls++;
              const summary =
                c.name === "Bash"
                  ? String(c.input?.command ?? "").slice(0, 200)
                  : c.name === "Write" || c.name === "Edit" || c.name === "Read"
                    ? String(c.input?.file_path ?? "")
                    : c.name?.startsWith("mcp__")
                      ? JSON.stringify(c.input).slice(0, 200)
                      : "";
              this.emit("tool.use", `${c.name}: ${summary}`, { tool: c.name }, "debug", spec.key);
            }
          }
        } else if (msg.type === "result") {
          const r = msg as unknown as {
            subtype: string;
            total_cost_usd?: number;
            num_turns?: number;
            result?: string;
            usage?: Record<string, unknown>;
            modelUsage?: Record<string, unknown>;
            errors?: string[];
          };
          result.subtype = r.subtype;
          result.costUsd = Number(r.total_cost_usd ?? 0);
          result.turns = Number(r.num_turns ?? 0);
          result.resultText = (r.result ?? (r.errors ?? []).join("; ")).slice(0, 4000);
          const u = (r.usage ?? {}) as Record<string, number>;
          result.usage = {
            model: spec.reviewer ? this.job.manifest.reviewerModel : this.job.manifest.model,
            inputTokens: u.input_tokens ?? 0,
            outputTokens: u.output_tokens ?? 0,
            cacheReadTokens: u.cache_read_input_tokens ?? 0,
            cacheCreateTokens: u.cache_creation_input_tokens ?? 0,
            modelUsage: r.modelUsage,
            toolCalls,
          };
        }
      }
    } catch (e) {
      result.subtype = abort.signal.aborted
        ? `aborted:${(abort.signal.reason as Error)?.message ?? "abort"}`
        : `error:${(e as Error).message.slice(0, 300)}`;
    } finally {
      clearTimeout(timer);
      clearInterval(cancelPoll);
    }
    result.denied = denied.n;
    await this.killLeftoverProcesses(spec.key);
    this.spent += result.costUsd;
    this.turns += result.turns;
    this.emit(
      "session.end",
      `${spec.key} session ${result.subtype}: ${result.turns} turns, $${result.costUsd.toFixed(3)}, ${Math.round((Date.now() - t0) / 1000)}s, ${denied.n} policy denials`,
      { subtype: result.subtype, costUsd: result.costUsd, turns: result.turns, denied: denied.n },
      "info",
      spec.key,
    );
    return result;
  }

  /** Kills processes (owned by this user) whose command line references this workspace — leftover solvers, servers, watchers. */
  private async killLeftoverProcesses(stage: StageKey): Promise<void> {
    try {
      const out = await new Promise<string>((resolve) => {
        const ps = spawn("/bin/ps", ["-axo", "pid=,command="], { stdio: ["ignore", "pipe", "ignore"] });
        let buf = "";
        ps.stdout.on("data", (d) => {
          buf += d.toString();
        });
        ps.on("close", () => resolve(buf));
        ps.on("error", () => resolve(""));
      });
      const mine = out
        .split("\n")
        .map((l) => l.trim())
        .filter((l) => l.includes(this.ws) && !/claude-agent-sdk|dist\/main\.js/.test(l))
        .map((l) => Number(l.split(/\s+/)[0]))
        .filter((pid) => Number.isInteger(pid) && pid !== process.pid);
      for (const pid of mine) {
        try {
          process.kill(pid, "SIGTERM");
        } catch {
          /* gone */
        }
      }
      if (mine.length) {
        await new Promise((r) => setTimeout(r, 3000));
        for (const pid of mine) {
          try {
            process.kill(pid, "SIGKILL");
          } catch {
            /* gone */
          }
        }
        this.emit(
          "process.cleanup",
          `terminated ${mine.length} leftover process(es) after the ${stage} session`,
          { pids: mine.slice(0, 20) },
          "warn",
          stage,
        );
      }
    } catch {
      /* best effort */
    }
  }

  private async stageContext(extra: Partial<StageContext> = {}): Promise<StageContext> {
    return {
      filesPresent: await listFiles(this.ws),
      previousStageNotes: this.notes.slice(-6).join("\n"),
      ...extra,
    };
  }

  private async syncAndCheckpoint(
    stage: StageKey,
    status: "completed" | "failed",
    cost: number,
    turns: number,
    usage?: Record<string, unknown>,
    error?: string,
  ) {
    await syncArtifacts(this.ws, stage, this.host.upload, (k, m, d) => this.emit(k, m, d, "info", stage));
    if (status === "completed") this.completed.push(stage);
    await this.host.stage({
      stage,
      status,
      costUsd: cost,
      turns,
      usage,
      error,
      checkpoint:
        status === "completed"
          ? { stage, completedStages: this.completed, spentUsd: this.spent, at: new Date().toISOString() }
          : undefined,
    });
  }

  /** Platform-side reproduction (stage 8, before the agent's revision session). */
  private async reproduce(): Promise<Record<string, unknown>> {
    const out: Record<string, unknown> = { at: new Date().toISOString() };
    const manuscript = join(this.ws, "manuscript");
    const cleanDir = join(this.ws, "_repro/manuscript");
    await rm(join(this.ws, "_repro"), { recursive: true, force: true });
    await mkdir(cleanDir, { recursive: true });
    for (const f of await readdir(manuscript))
      if (/\.(tex|bib|cls|sty|bst|pdf|png|jpg|csv)$/.test(f) && f !== "main.pdf")
        await cp(join(manuscript, f), join(cleanDir, f), { recursive: true });
    for (const d of ["results"])
      if (await stat(join(this.ws, d)).catch(() => null))
        await cp(join(this.ws, d), join(this.ws, "_repro", d), { recursive: true });
    const build = await buildLatex(cleanDir, {
      tectonicBin: this.host.tectonicBin,
      cacheDir: this.host.tectonicCache,
    });
    out.cleanPdfBuild = {
      ok: build.ok,
      pages: build.pages,
      durationMs: build.durationMs,
      logTail: build.log.slice(-1500),
    };
    if (build.ok && build.pdfPath) {
      const a = createHash("sha256")
        .update(await readFile(build.pdfPath))
        .digest("hex");
      out.cleanPdfSha256 = a;
    }
    const script = join(this.ws, "code/reproduce.sh");
    if (await stat(script).catch(() => null)) {
      const reproDir = join(this.ws, "_repro/work");
      await mkdir(reproDir, { recursive: true });
      for (const d of ["code", "data"])
        if (await stat(join(this.ws, d)).catch(() => null))
          await cp(join(this.ws, d), join(reproDir, d), {
            recursive: true,
            filter: (s) => !/\/(\.venv|node_modules|__pycache__)(\/|$)/.test(s),
          });
      await mkdir(join(reproDir, "results/tables"), { recursive: true });
      await mkdir(join(reproDir, "results/figures"), { recursive: true });
      await mkdir(join(reproDir, "home"), { recursive: true });
      // Seatbelt matches canonical paths (/private/var/...): resolve every path and allow both spellings
      const real = async (p: string) => realpath(p).catch(() => p);
      const reproReal = await real(reproDir);
      const wsReal = await real(this.ws);
      const cacheDir = this.host.tectonicCache ?? join(this.ws, ".cache/tectonic");
      const cacheReal = await real(cacheDir);
      const both = (p: string, q: string) => [...new Set([p, q])].map((x) => `(subpath "${x}")`).join(" ");
      const profile = [
        "(version 1)",
        "(allow default)",
        "(deny file-write*)",
        `(allow file-write* ${both(reproDir, reproReal)} (subpath "/private/tmp") (subpath "/tmp") (subpath "/private/var/folders") (literal "/dev/null") (regex #"^/dev/tty") ${both(cacheDir, cacheReal)} ${both(join(this.ws, ".cache"), join(wsReal, ".cache"))})`,
        `(deny file-read* (subpath "/Users") (subpath "/etc/autoscholar") (subpath "/private/etc/autoscholar") (subpath "/var/autoscholar/runs") (subpath "/private/var/autoscholar/runs"))`,
        `(allow file-read* ${both(reproDir, reproReal)} ${both(join(this.ws, ".cache"), join(wsReal, ".cache"))} ${both(cacheDir, cacheReal)})`,
        "",
      ].join("\n");
      const profilePath = join(this.ws, "_repro/repro.sb");
      await writeFile(profilePath, profile);
      const t0 = Date.now();
      const res = await new Promise<{ code: number | null; log: string }>((resolve) => {
        const child = spawn("/usr/bin/sandbox-exec", ["-f", profilePath, "/bin/bash", "code/reproduce.sh"], {
          cwd: reproDir,
          env: {
            PATH: "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin",
            HOME: join(reproDir, "home"),
            TMPDIR: "/tmp",
            UV_CACHE_DIR: join(this.ws, ".cache/uv"),
            MPLBACKEND: "Agg",
            LANG: "en_US.UTF-8",
            PYTHONDONTWRITEBYTECODE: "1",
          },
          stdio: ["ignore", "pipe", "pipe"],
        });
        let log = "";
        const cap = (d: Buffer) => {
          log += d.toString();
          if (log.length > 200_000) log = log.slice(-150_000);
        };
        child.stdout.on("data", cap);
        child.stderr.on("data", cap);
        const timer = setTimeout(() => child.kill("SIGKILL"), 90 * 60_000);
        child.on("close", (code) => {
          clearTimeout(timer);
          resolve({ code, log });
        });
        child.on("error", (e) => resolve({ code: -1, log: `spawn error: ${e.message}` }));
      });
      out.reproduceScript = {
        ran: res.code !== -1,
        exitCode: res.code,
        durationMs: Date.now() - t0,
        logTail: res.log.slice(-3000),
      };
      try {
        const orig = MetricsFile.parse(
          JSON.parse(await readFile(join(this.ws, "results/metrics.json"), "utf8")),
        );
        const regen = MetricsFile.parse(
          JSON.parse(await readFile(join(reproDir, "results/metrics.json"), "utf8")),
        );
        const compared: Array<{ key: string; original: unknown; reproduced: unknown; match: boolean }> = [];
        for (const [k, v] of Object.entries(orig.metrics)) {
          const r = regen.metrics[k];
          const match =
            typeof v === "number" && typeof r === "number"
              ? Math.abs(v - r) <= Math.max(1e-9, Math.abs(v) * 1e-6)
              : JSON.stringify(v) === JSON.stringify(r);
          compared.push({ key: k, original: v, reproduced: r, match });
        }
        out.metricsCompared = compared.length;
        out.metricsMismatched = compared.filter((c) => !c.match).slice(0, 25);
        out.metricsMissingInRerun = compared.filter((c) => c.reproduced === undefined).length;
      } catch (e) {
        out.metricsComparison = `not compared: ${(e as Error).message.slice(0, 200)}`;
      }
    } else out.reproduceScript = { ran: false, reason: "code/reproduce.sh not present" };
    await mkdir(join(this.ws, "results"), { recursive: true });
    await writeFile(join(this.ws, "results/reproduction.json"), `${JSON.stringify(out, null, 2)}\n`);
    return out;
  }

  async run(): Promise<RunOutcome> {
    const { startIndex } = await this.prepareWorkspace();
    const finish = async (outcome: RunOutcome["outcome"], error?: string): Promise<RunOutcome> => {
      await this.finalize(outcome, error);
      return {
        outcome,
        error,
        costUsd: this.spent,
        turns: this.turns,
        stagesCompleted: this.completed,
        resourceUsage: {
          wallSeconds: (Date.now() - this.started) / 1000,
          diskBytes: await dirSize(this.ws),
          peakRssBytes: this.peakRss,
        },
      };
    };
    const total = this.job.manifest.maxBudgetUsd;
    if (startIndex >= STAGE_SPECS.length) {
      const r = await this.qaRevision();
      if (r) return finish(r.outcome, r.error);
    }
    for (let i = startIndex; i < STAGE_SPECS.length; i++) {
      const spec = STAGE_SPECS[i]!;
      if (this.host.isCancelled()) return finish("cancelled", "cancellation requested");
      if (this.remainingBudget() < Math.max(0.25, total * 0.02))
        return finish(
          "budget_exceeded",
          `budget exhausted before ${spec.key} ($${this.spent.toFixed(2)} of $${total})`,
        );
      if (this.remainingMinutes() < 2) return finish("timeout", `runtime cap reached before ${spec.key}`);
      await this.host.stage({ stage: spec.key, status: "running", costUsd: 0, turns: 0 });
      this.emit("stage.start", `stage ${i}: ${spec.key}`, undefined, "info", spec.key);
      const stageBudget = Math.min(this.remainingBudget() * 0.9, Math.max(1, total * spec.budgetShare * 1.6));
      const stageMinutes = Math.min(
        this.remainingMinutes() * 0.9,
        Math.max(5, this.job.manifest.maxRuntimeMinutes * spec.wallMinutesShare * 1.8),
      );
      let ctx: StageContext;
      if (spec.key === "reproduction") {
        this.emit(
          "reproduction.start",
          "platform re-executing analysis and clean LaTeX build",
          undefined,
          "info",
          spec.key,
        );
        const repro = await this.reproduce();
        const pdf = join(this.ws, "manuscript/main.pdf");
        let previews: string[] = [];
        if (await stat(pdf).catch(() => null)) {
          const pages = Number((repro.cleanPdfBuild as { pages?: number })?.pages ?? 0) || 4;
          const picks = [...new Set([1, 2, Math.max(1, Math.floor(pages / 2)), pages])].filter(
            (p) => p >= 1 && p <= pages,
          );
          previews = await renderPages(pdf, join(this.ws, "review/previews"), picks);
          this.emit(
            "reproduction.previews",
            previews.length
              ? `rendered ${previews.length} page previews`
              : "page previews unavailable (pdftoppm missing)",
            { previews: previews.map((p) => p.replace(`${this.ws}/`, "")) },
            "info",
            spec.key,
          );
        }
        ctx = await this.stageContext({
          reproduction: repro,
          previewFiles: previews.map((p) => p.replace(`${this.ws}/`, "")),
        });
      } else ctx = await this.stageContext();
      let res = await this.session(spec, spec.prompt(this.job.manifest, ctx), {
        maxBudgetUsd: stageBudget,
        maxTurns: spec.maxTurns,
        timeoutMs: stageMinutes * 60_000,
      });
      if (res.subtype.startsWith("aborted:cancelled"))
        return finish("cancelled", `cancelled during ${spec.key}`);
      if (res.subtype === "error_max_budget_usd" && this.remainingBudget() < 1) {
        await this.syncAndCheckpoint(
          spec.key,
          "failed",
          res.costUsd,
          res.turns,
          res.usage,
          "budget exhausted",
        );
        return finish("budget_exceeded", `budget exhausted during ${spec.key}`);
      }
      // manuscript stage: the platform guarantees a PDF build attempt
      if (spec.key === "manuscript" || spec.key === "reproduction") {
        const b = await buildLatex(join(this.ws, "manuscript"), {
          tectonicBin: this.host.tectonicBin,
          cacheDir: this.host.tectonicCache,
        });
        this.emit(
          "latex.build",
          b.ok ? `platform build ok: ${b.pages} pages` : "platform build failed",
          { ok: b.ok, pages: b.pages, log: b.log.slice(-1200) },
          b.ok ? "info" : "warn",
          spec.key,
        );
        if (!b.ok && this.remainingBudget() > 1 && !this.host.isCancelled()) {
          const fix = await this.session(
            spec,
            `${spec.prompt(this.job.manifest, await this.stageContext())}\n\n# BUILD FAILED — fix it\nThe platform's clean tectonic build of manuscript/main.tex failed. Log:\n\`\`\`\n${b.log.slice(-6000)}\n\`\`\`\nFix the LaTeX (or bibliography) and call latex_build until it succeeds. Do not remove scientific content to make it compile.`,
            { maxBudgetUsd: Math.min(this.remainingBudget() * 0.5, 4), maxTurns: 60, timeoutMs: 20 * 60_000 },
          );
          res = {
            ...res,
            costUsd: res.costUsd + fix.costUsd,
            turns: res.turns + fix.turns,
            denied: res.denied + fix.denied,
          };
          const b2 = await buildLatex(join(this.ws, "manuscript"), {
            tectonicBin: this.host.tectonicBin,
            cacheDir: this.host.tectonicCache,
          });
          this.emit(
            "latex.build",
            b2.ok ? `rebuild ok: ${b2.pages} pages` : "rebuild failed",
            { ok: b2.ok, pages: b2.pages },
            b2.ok ? "info" : "error",
            spec.key,
          );
          if (!b2.ok) {
            await this.syncAndCheckpoint(
              spec.key,
              "failed",
              res.costUsd,
              res.turns,
              res.usage,
              "LaTeX build failed after fix attempt",
            );
            return finish("failed", `LaTeX build failed in ${spec.key}`);
          }
        } else if (!b.ok) {
          await this.syncAndCheckpoint(
            spec.key,
            "failed",
            res.costUsd,
            res.turns,
            res.usage,
            "LaTeX build failed",
          );
          return finish("failed", `LaTeX build failed in ${spec.key}`);
        }
      }
      let problems = await spec.check(this.ws, this.job.manifest);
      if (problems.some((p) => p.startsWith("INFEASIBLE"))) {
        await this.syncAndCheckpoint(
          spec.key,
          "failed",
          res.costUsd,
          res.turns,
          res.usage,
          problems.join("; "),
        );
        return finish(
          "failed",
          problems.find((p) => p.startsWith("INFEASIBLE")),
        );
      }
      if (problems.length && this.remainingBudget() > 1 && !this.host.isCancelled()) {
        this.emit(
          "stage.check",
          `post-conditions not met, running a fix session: ${problems.join("; ")}`,
          { problems },
          "warn",
          spec.key,
        );
        const fix = await this.session(
          spec,
          `${spec.prompt(this.job.manifest, await this.stageContext())}\n\n# INCOMPLETE — the deterministic post-conditions of this stage failed:\n- ${problems.join("\n- ")}\nComplete exactly what is missing (do not redo finished work).`,
          {
            maxBudgetUsd: Math.min(this.remainingBudget() * 0.5, Math.max(1, stageBudget * 0.5)),
            maxTurns: Math.ceil(spec.maxTurns / 2),
            timeoutMs: Math.max(5, stageMinutes * 0.5) * 60_000,
          },
        );
        res = {
          ...res,
          costUsd: res.costUsd + fix.costUsd,
          turns: res.turns + fix.turns,
          denied: res.denied + fix.denied,
        };
        if (fix.subtype.startsWith("aborted:cancelled"))
          return finish("cancelled", `cancelled during ${spec.key}`);
        problems = await spec.check(this.ws, this.job.manifest);
      }
      if (problems.length) {
        await this.syncAndCheckpoint(
          spec.key,
          "failed",
          res.costUsd,
          res.turns,
          res.usage,
          problems.join("; "),
        );
        return finish("failed", `${spec.key}: ${problems.join("; ")}`);
      }
      this.notes.push(`[${spec.key}] ${res.resultText.replace(/\s+/g, " ").slice(0, 400)}`);
      await this.syncAndCheckpoint(spec.key, "completed", res.costUsd, res.turns, res.usage);
    }
    return finish("completed");
  }

  /**
   * Retry after `quality_failed`: fetch the server's QA report, run one revision session that addresses the hard
   * failures (same rules as the reproduction stage), rebuild, and let finalize/QA re-run. Returns an outcome only on failure.
   */
  private async qaRevision(): Promise<{ outcome: RunOutcome["outcome"]; error?: string } | null> {
    const spec = STAGE_SPECS.find((s) => s.key === "reproduction")!;
    let report: { hardFailures?: string[]; warnings?: string[] } | null = null;
    try {
      const list = (await this.host.fetchArtifacts?.()) ?? [];
      const qa = list.find((a) => a.path === "review/qa-report.json");
      if (qa) {
        const buf = await qa.read();
        await mkdir(join(this.ws, "review"), { recursive: true });
        await writeFile(join(this.ws, "review/qa-report.json"), buf);
        report = JSON.parse(buf.toString("utf8"));
      }
    } catch (e) {
      this.emit(
        "qa.revision",
        `could not fetch the QA report: ${(e as Error).message}`,
        undefined,
        "warn",
        "reproduction",
      );
    }
    const hard = report?.hardFailures ?? [];
    if (hard.length === 0) {
      this.emit(
        "qa.revision",
        "no QA report with hard failures; re-finalizing only",
        undefined,
        "info",
        "reproduction",
      );
      return null;
    }
    if (this.remainingBudget() < 2)
      return { outcome: "budget_exceeded", error: "no budget left for the QA revision" };
    await this.host.stage({ stage: "reproduction", status: "running", costUsd: 0, turns: 0 });
    this.emit(
      "qa.revision",
      `revision pass for ${hard.length} failed hard gate(s)`,
      { hardFailures: hard },
      "info",
      "reproduction",
    );
    const prompt = `${spec.prompt(this.job.manifest, await this.stageContext())}

# QUALITY GATES FAILED — revision pass
The platform's deterministic publication gates rejected the previous attempt. Fix the causes in the artifacts (never by weakening claims you cannot support; downgrade or remove unsupported claims instead), then rebuild with latex_build. Failed hard gates:
- ${hard.join("\n- ")}
Soft warnings: ${(report?.warnings ?? []).join(" | ") || "none"}.
Contract reminders: results/metrics.json must be {"metrics": {...}, "reportedValues": [...]} (register every reported number with record_reported_value); results/experiment-log.jsonl entries need "name", "command", "startedAt", "endedAt", "exitCode"; code/reproduce.sh must finish within 60 minutes (put long certificate re-checks behind an optional flag and say so in the paper); review/review-1.json must keep every finding with a resolution.`;
    const res = await this.session(spec, prompt, {
      maxBudgetUsd: Math.min(this.remainingBudget() * 0.6, 20),
      maxTurns: 120,
      timeoutMs: 60 * 60_000,
    });
    if (res.subtype.startsWith("aborted:cancelled"))
      return { outcome: "cancelled", error: "cancelled during QA revision" };
    const b = await buildLatex(join(this.ws, "manuscript"), {
      tectonicBin: this.host.tectonicBin,
      cacheDir: this.host.tectonicCache,
    });
    this.emit(
      "latex.build",
      b.ok ? `revision build ok: ${b.pages} pages` : "revision build failed",
      { ok: b.ok, pages: b.pages },
      b.ok ? "info" : "error",
      "reproduction",
    );
    const problems = await spec.check(this.ws, this.job.manifest);
    await this.syncAndCheckpoint(
      "reproduction",
      problems.length || !b.ok ? "failed" : "completed",
      res.costUsd,
      res.turns,
      res.usage,
      problems.join("; ") || (b.ok ? undefined : "build failed"),
    );
    if (!b.ok) return { outcome: "failed", error: "LaTeX build failed after QA revision" };
    if (problems.length) return { outcome: "failed", error: `QA revision: ${problems.join("; ")}` };
    return null;
  }

  private async finalize(outcome: RunOutcome["outcome"], error?: string) {
    const summary = {
      runId: this.job.runId,
      attemptNo: this.job.attemptNo,
      researchMode: this.job.manifest.researchMode,
      stagesCompleted: this.completed,
      costUsd: Number(this.spent.toFixed(4)),
      turns: this.turns,
      model: this.job.manifest.model,
      reviewerModel: this.job.manifest.reviewerModel,
      workerName: this.host.workerName,
      startedAt: new Date(this.started).toISOString(),
      endedAt: new Date().toISOString(),
      sdkVersion: this.host.sdkVersion,
      toolchain: this.host.toolchain,
      synthetic: false,
      outcome,
      error: error ?? null,
    };
    const step = async (name: string, fn: () => Promise<unknown>) => {
      try {
        await fn();
      } catch (e) {
        this.emit(
          "run.finalize_error",
          `${name}: ${(e as Error).message}`.slice(0, 1000),
          undefined,
          "error",
        );
      }
    };
    await step("cleanup-processes", () => this.killLeftoverProcesses("reproduction"));
    await step("summary", async () => {
      await mkdir(join(this.ws, "provenance"), { recursive: true });
      await writeFile(join(this.ws, "provenance/run-summary.json"), `${JSON.stringify(summary, null, 2)}\n`);
    });
    await step("cleanup", () => rm(join(this.ws, "_repro"), { recursive: true, force: true }));
    // sync first, then derive checksums.sha256 from the digests that were actually uploaded, then upload it
    await step("sync", () =>
      syncArtifacts(this.ws, "reproduction", this.host.upload, (k, m, d) => this.emit(k, m, d)),
    );
    await step("checksums", async () => {
      const text = await writeChecksums(this.ws, join(this.ws, ".autoscholar/uploaded.json"));
      await this.host.upload(
        "provenance/checksums.sha256",
        "checksum",
        "text/plain; charset=utf-8",
        Buffer.from(text),
      );
    });
    this.emit(
      "run.finalized",
      `run ${outcome}${error ? `: ${error}` : ""} — $${this.spent.toFixed(3)}, ${this.turns} turns, ${this.completed.length} stages`,
      { outcome, costUsd: this.spent, turns: this.turns },
      outcome === "completed" ? "info" : "warn",
    );
  }
}

export async function runResearch(job: RunnerJob, host: RunnerHost): Promise<RunOutcome> {
  return new ResearchRunner(job, host).run();
}