import { describe, expect, test } from "bun:test";
import { readFileSync } from "node:fs";
import { mkdtemp, rm } from "node:fs/promises";
import { createServer } from "node:net";
import { tmpdir } from "node:os";
import { join } from "node:path";
import {
  createClusterAdapter,
  parsePsOutput,
  queryAgentGuard,
  type AgentGuardPollResponse,
} from "./cluster";

const FIXTURE_DIR = join(import.meta.dir, "../../test/fixtures/cluster");
const AGENT_CLASS_REGEX = "(claude|cursor-agent|ccr|harnessd|dead-advisor)";
const NOW = 1_700_000_000_000;

function fixture(name: string): string {
  return readFileSync(join(FIXTURE_DIR, name), "utf8");
}

function guardFixture(): AgentGuardPollResponse {
  return JSON.parse(fixture("agent-guard-poll.json")) as AgentGuardPollResponse;
}

interface HarnessOptions {
  buildboxState?: string;
  ciFallbackDir?: string;
  psFile?: string;
  guard?: AgentGuardPollResponse;
  guardError?: Error;
  psError?: Error;
  statImpl?: (path: string) => { size: number; mtimeMs: number };
  readFileImpl?: (path: string) => string;
  guardSocketPath?: string;
}

function guardQuery(opts: HarnessOptions): (socketPath: string) => Promise<AgentGuardPollResponse> {
  if (opts.guardSocketPath) return queryAgentGuard;
  const guardError = opts.guardError;
  if (guardError) {
    return async () => {
      throw guardError;
    };
  }
  return async () => opts.guard ?? guardFixture();
}

function harness(opts: HarnessOptions = {}) {
  const buildboxState = opts.buildboxState ?? "buildbox-state-online.txt";
  const ciFallbackDir = opts.ciFallbackDir ?? "ci-fallback-pressure";
  const psFile = opts.psFile ?? "ps-with-orphan.txt";

  return createClusterAdapter({
    agentClassRegex: AGENT_CLASS_REGEX,
    now: () => NOW,
    buildboxStatePath: join(FIXTURE_DIR, buildboxState),
    ciFallbackDir: join(FIXTURE_DIR, ciFallbackDir),
    buildslotDir: join(FIXTURE_DIR, "buildslot"),
    agentGuardSocketPath: opts.guardSocketPath,
    queryAgentGuard: guardQuery(opts),
    runPs: opts.psError
      ? async () => {
          throw opts.psError;
        }
      : async () => fixture(psFile),
    readFileImpl: opts.readFileImpl ?? ((path: string) => {
      if (path.startsWith("/proc/") && path.endsWith("/cgroup")) {
        const pid = path.split("/")[2];
        if (pid === "88231") {
          return "0::/user.slice/user-1000.slice/session-3.scope/app.slice/agent.scope\n";
        }
        if (pid === "44100" || pid === "44101") {
          return "0::/user.slice/user-1000.slice/session-3.scope/app.slice/claude.scope\n";
        }
      }
      return readFileSync(path, "utf8");
    }),
    statImpl:
      opts.statImpl ??
      ((path: string) => {
        if (path.endsWith("slot-0.lock")) return { size: 128, mtimeMs: NOW - 60_000 };
        if (path.endsWith("slot-1.lock")) return { size: 0, mtimeMs: NOW - 60_000 };
        if (path.endsWith("ticket.aaaaaa.lock")) return { size: 0, mtimeMs: NOW - 60_000 };
        if (path.endsWith("ticket.bbbbbb.lock")) return { size: 0, mtimeMs: NOW - 120_000 };
        if (path.endsWith("ticket.cccccc.lock")) return { size: 0, mtimeMs: NOW - 300_000 };
        return { size: 0, mtimeMs: NOW };
      }),
  });
}

describe("createClusterAdapter", () => {
  test("speaks the agent-guard poll protocol over a unix socket", async () => {
    const dir = await mkdtemp(join(tmpdir(), "agent-guard-poll-"));
    const socketPath = join(dir, "agent-guard.sock");
    const requests: string[] = [];
    const server = createServer((socket) => {
      socket.once("data", (data) => {
        requests.push(data.toString().trim());
        socket.end(JSON.stringify(guardFixture()));
      });
    });
    await new Promise<void>((resolve, reject) => {
      server.once("error", reject);
      server.listen(socketPath, resolve);
    });

    try {
      const agents = (await harness({ guardSocketPath: socketPath }).poll()).panels.find(
        (panel) => panel.id === "agents",
      )?.data as {
        events: AgentGuardPollResponse["events"];
        culprits: AgentGuardPollResponse["culprits"];
      };
      expect(requests).toEqual([JSON.stringify({ type: "poll" })]);
      expect(agents.events).toEqual(guardFixture().events);
      expect(agents.culprits).toEqual(guardFixture().culprits);
    } finally {
      await new Promise<void>((resolve) => server.close(() => resolve()));
      await rm(dir, { recursive: true, force: true });
    }
  });

  test("emits cluster and agents panels from fixture sources", async () => {
    const result = await harness().poll();

    expect(result.panels.map((p) => p.id).sort()).toEqual(["agents", "cluster"]);

    const cluster = result.panels.find((p) => p.id === "cluster")?.data as {
      debian1: string;
      autoscaler: { state: string; pressure: number };
      buildslot: { running: number; queued: number; p95WaitSeconds: number | null };
    };
    expect(cluster.debian1).toBe("online");
    expect(cluster.autoscaler.state).toBe("pressure");
    expect(cluster.autoscaler.pressure).toBe(2);
    expect(cluster.buildslot.running).toBe(1);
    expect(cluster.buildslot.queued).toBe(3);
    expect(cluster.buildslot.p95WaitSeconds).toBe(300);

    const agents = result.panels.find((p) => p.id === "agents")?.data as {
      fleet: Array<{ class: string; count: number; cpuPercent: number; slices: string[] }>;
      orphans: number;
      events: unknown[];
      culprits: unknown[];
      totalLive: number;
    };
    expect(agents.totalLive).toBe(3);
    expect(agents.orphans).toBe(1);
    expect(agents.events).toHaveLength(1);
    expect(agents.culprits).toHaveLength(1);
    expect(agents.fleet).toEqual(
      expect.arrayContaining([
        expect.objectContaining({ class: "cursor-agent", count: 1, slices: ["app.slice"] }),
        expect.objectContaining({ class: "claude", count: 2, slices: ["app.slice"] }),
      ]),
    );
  });

  test("debian1 offline is reflected in the cluster panel", async () => {
    const result = await harness({ buildboxState: "buildbox-state-offline.txt" }).poll();
    const cluster = result.panels.find((p) => p.id === "cluster")?.data as { debian1: string };
    expect(cluster.debian1).toBe("offline");
  });

  test("autoscaler idle state comes from ci-fallback counters", async () => {
    const result = await harness({ ciFallbackDir: "ci-fallback-idle" }).poll();
    const cluster = result.panels.find((p) => p.id === "cluster")?.data as {
      autoscaler: { state: string; idle: number };
    };
    expect(cluster.autoscaler.state).toBe("idle");
    expect(cluster.autoscaler.idle).toBe(3);
  });

  test("orphan candidate emits act alert with reap action", async () => {
    const result = await harness().poll();
    const alerts = result.items.filter((i) => i.kind === "alert");
    expect(alerts).toHaveLength(1);
    expect(alerts[0]?.id).toBe("cluster:orphan:88231");
    expect(alerts[0]?.source).toBe("cluster");
    expect(alerts[0]?.severity).toBe("act");
    expect(alerts[0]?.title).toContain("94% CPU");
    expect(alerts[0]?.title).toContain("3h12m");
    expect(alerts[0]?.detail).toContain("cursor-agent 88231");
    expect(alerts[0]?.actions).toEqual([
      { verb: "reap", args: { pid: "88231" }, label: "Reap", recommended: true },
    ]);
  });

  test("no orphan items when ps output has no qualifying processes", async () => {
    const result = await harness({ psFile: "ps-no-orphans.txt" }).poll();
    expect(result.items).toEqual([]);
    const agents = result.panels.find((p) => p.id === "agents")?.data as { orphans: number };
    expect(agents.orphans).toBe(0);
  });

  test("etime and cpu boundaries are exclusive: at 30m and 80% does not fire", async () => {
    const result = await harness({ psFile: "ps-boundary-clear.txt" }).poll();
    expect(result.items).toEqual([]);
  });

  test("declarative snapshot: orphan absent on next poll when ps clears", async () => {
    const adapter = harness();
    const first = await adapter.poll();
    expect(first.items).toHaveLength(1);

    const cleared = harness({ psFile: "ps-no-orphans.txt" });
    const second = await cleared.poll();
    expect(second.items).toEqual([]);
  });

  test("agent-guard socket failure rejects the poll", async () => {
    const adapter = harness({ guardError: new Error("connect ENOENT agent-guard.sock") });
    await expect(adapter.poll()).rejects.toThrow("agent-guard");
  });

  test("ps failure rejects the poll", async () => {
    const adapter = harness({ psError: new Error("rtk proxy ps failed") });
    await expect(adapter.poll()).rejects.toThrow("rtk");
  });

  test("invalid buildbox state rejects the poll", async () => {
    const adapter = createClusterAdapter({
      agentClassRegex: AGENT_CLASS_REGEX,
      buildboxStatePath: join(FIXTURE_DIR, "buildbox-state-online.txt"),
      ciFallbackDir: join(FIXTURE_DIR, "ci-fallback-pressure"),
      buildslotDir: join(FIXTURE_DIR, "buildslot"),
      queryAgentGuard: async () => guardFixture(),
      runPs: async () => fixture("ps-no-orphans.txt"),
      readFileImpl: (path) => {
        if (path.endsWith("buildbox-state-online.txt")) return "unknown";
        return readFileSync(path, "utf8");
      },
    });
    await expect(adapter.poll()).rejects.toThrow("buildbox-watch state");
  });

  test("unreadable ci-fallback counter rejects the poll", async () => {
    const adapter = harness({ ciFallbackDir: "ci-fallback-missing" });
    await expect(adapter.poll()).rejects.toThrow();
  });

  test("reads only configured ci-fallback idle and pressure sources", async () => {
    const reads: string[] = [];
    const adapter = harness({
      readFileImpl: (path) => {
        reads.push(path);
        if (path.startsWith("/proc/")) return "0::/app.slice/agent.scope\n";
        return readFileSync(path, "utf8");
      },
    });
    await adapter.poll();
    expect(reads.some((path) => path.endsWith("/redundant"))).toBe(false);
  });

  test("unreadable queued buildslot ticket rejects the poll", async () => {
    const adapter = harness({
      statImpl: (path) => {
        if (path.endsWith("slot-0.lock")) return { size: 128, mtimeMs: NOW - 60_000 };
        if (path.endsWith("slot-1.lock")) return { size: 0, mtimeMs: NOW - 60_000 };
        throw new Error(`cannot stat ${path}`);
      },
    });
    await expect(adapter.poll()).rejects.toThrow("cannot stat");
  });

  test("unreadable agent cgroup rejects the poll", async () => {
    const adapter = harness({
      readFileImpl: (path) => {
        if (path.startsWith("/proc/")) throw new Error(`cannot read ${path}`);
        return readFileSync(path, "utf8");
      },
    });
    await expect(adapter.poll()).rejects.toThrow("cannot read /proc/");
  });
});

describe("parsePsOutput", () => {
  test("classifies agent processes with the configured regex", () => {
    const processes = parsePsOutput(fixture("ps-with-orphan.txt"), AGENT_CLASS_REGEX);
    expect(processes.filter((p) => p.agentClass).map((p) => p.agentClass)).toEqual([
      "cursor-agent",
      "claude",
      "claude",
    ]);
    expect(processes.find((p) => p.pid === 88231)?.etimeSeconds).toBe(11_565);
  });

  test("rejects malformed process rows instead of returning a partial snapshot", () => {
    expect(() => parsePsOutput("PID PPID NI %CPU ELAPSED COMMAND\n88231 broken", AGENT_CLASS_REGEX))
      .toThrow("invalid ps row");
  });

  test("accepts rtk ps rows with dash NI and day-prefixed etime", () => {
    const row = "1289643    8675   - 77.4    12:04:46 node src/control-api.js --port 4974";
    const processes = parsePsOutput(`PID PPID NI %CPU ELAPSED COMMAND\n${row}`, AGENT_CLASS_REGEX);
    expect(processes).toHaveLength(1);
    expect(processes[0]?.ni).toBe(0);
    expect(processes[0]?.etimeSeconds).toBe(12 * 3_600 + 4 * 60 + 46);
  });
});