import { execFile, spawn } from "node:child_process";
import { access } from "node:fs/promises";
import { promisify } from "node:util";
import type { UnitOfWork } from "@awp/application";
import { executionCallbackToken } from "@awp/application";
import {
  issueModelGatewayCapability,
  unsafeOpaqueId,
  type AgentRunId,
  type TaskId,
  type WorkspaceId,
} from "@awp/contracts";
import type {
  AcpNativeCapabilities,
  AcpNativeClient,
  AcpNativeSession,
  AcpFailureInjection,
  AcpAttemptSelectionKind,
} from "@awp/provider-agent-acp";
import { resourceBaseName } from "@awp/provider-workspace-kubernetes";

const execFileAsync = promisify(execFile);
const RUNNER_PATH = "/opt/awp-agent-runner/runner.mjs";

export interface KubernetesAcpNativeClientOptions {
  readonly namespace: string;
  readonly callbackBaseUrl: string;
  readonly callbackSecret: string;
  readonly modelGatewayBaseUrl: string;
  readonly modelGatewaySigningSecret: string;
  readonly sourceRepositoryPath?: string;
  readonly selfRepositoryUrl?: string;
  readonly kubectlCommand?: string;
  readonly readyTimeoutMs?: number;
}

interface RunnerEnvelopeSession extends AcpNativeSession {
  readonly recoverable?: boolean;
}

function normalizeRepositoryUrl(value: string): string {
  return value
    .trim()
    .replace(/\.git$/, "")
    .replace(/\/$/, "")
    .toLowerCase();
}

export class KubernetesAcpNativeClient implements AcpNativeClient {
  private readonly kubectl: string;
  private readonly readyTimeoutMs: number;

  constructor(
    private readonly uow: UnitOfWork,
    private readonly options: KubernetesAcpNativeClientOptions,
  ) {
    this.kubectl = options.kubectlCommand ?? "kubectl";
    this.readyTimeoutMs = options.readyTimeoutMs ?? 120_000;
    if (!options.namespace.trim()) throw new Error("ACP Kubernetes namespace must not be empty");
    if (!options.callbackSecret.trim()) throw new Error("ACP callback secret must not be empty");
    if (!options.modelGatewaySigningSecret.trim()) {
      throw new Error("ACP model gateway signing secret must not be empty");
    }
  }

  async capabilities(): Promise<AcpNativeCapabilities> {
    return {
      protocolVersion: "1",
      agentName: "codex-acp",
      capabilities: [
        "session.new",
        "session.load",
        "session.resume",
        "session.cancel",
        "tool-call",
      ],
    };
  }

  async findSessionByIdempotencyKey(
    workspaceId: string,
    idempotencyKey: string,
  ): Promise<AcpNativeSession | undefined> {
    await this.waitForRunner(workspaceId);
    const response = await this.runnerRequest(workspaceId, {
      action: "find",
      idempotencyKey,
    });
    return response === null ? undefined : this.session(response);
  }

  async startSession(input: {
    readonly attemptId: string;
    readonly agentRunId: string;
    readonly taskId: string;
    readonly workspaceId: string;
    readonly providerId: string;
    readonly accountId: string;
    readonly model: string;
    readonly idempotencyKey: string;
    readonly correlationId: string;
    readonly selectionKind?: AcpAttemptSelectionKind;
    readonly failureInjection?: AcpFailureInjection;
  }): Promise<AcpNativeSession> {
    const launch = await this.resolveLaunch(input.agentRunId, input.taskId);
    await this.waitForRunner(input.workspaceId);
    const baseRevision = await this.ensureRepository(
      input.workspaceId,
      launch.repositoryUrl,
      launch.role === "reviewer" ? launch.baseRevision : undefined,
    );
    if (launch.role === "reviewer") {
      await this.materializeReviewCandidate(input.workspaceId, launch.diff, launch.candidateDigest);
    }
    const modelGatewayCapability = issueModelGatewayCapability(
      this.options.modelGatewaySigningSecret,
      { attemptId: input.attemptId, accountId: input.accountId },
    );
    const { failureInjection: _requestedFailureInjection, ...sessionInput } = input;
    const response = await this.runnerRequest(input.workspaceId, {
      action: "start",
      idempotencyKey: input.idempotencyKey,
      input: {
        ...sessionInput,
        ...(launch.role === "coder" && input.selectionKind === "initial"
          ? { failureInjection: "dogfood-post-prompt-failure-17" as const }
          : {}),
        role: launch.role,
        prompt: launch.prompt,
        baseRevision,
        modelGatewayBaseUrl: this.options.modelGatewayBaseUrl,
        modelGatewayCapability,
        ...(launch.role === "reviewer"
          ? {
              reviewId: launch.reviewId,
              candidateDigest: launch.candidateDigest,
              reviewCallbackUrl: `${this.options.callbackBaseUrl}/internal/execution/review`,
            }
          : { callbackUrl: `${this.options.callbackBaseUrl}/internal/execution/complete` }),
        failureUrl: `${this.options.callbackBaseUrl}/internal/execution/fail`,
        callbackToken: executionCallbackToken(this.options.callbackSecret, input.attemptId),
      },
    });
    if (response === null) throw new Error("ACP workspace runner returned no session");
    return this.session(response);
  }

  async cancelSession(nativeSessionId: string): Promise<AcpNativeSession> {
    const parsed = this.nativeSessionIdentity(nativeSessionId);
    const response = await this.runnerRequest(parsed.workspaceId, {
      action: "cancel",
      key: parsed.key,
    });
    if (response === null) throw Object.assign(new Error("ACP session not found"), { status: 404 });
    return this.session(response);
  }

  async getSession(nativeSessionId: string): Promise<AcpNativeSession | undefined> {
    const parsed = this.nativeSessionIdentity(nativeSessionId);
    const response = await this.runnerRequest(parsed.workspaceId, {
      action: "get",
      key: parsed.key,
    });
    return response === null ? undefined : this.session(response);
  }

  private async resolveLaunch(
    agentRunIdValue: string,
    taskIdValue: string,
  ): Promise<
    | { repositoryUrl: string; prompt: string; role: "coder" }
    | {
        repositoryUrl: string;
        prompt: string;
        role: "reviewer";
        reviewId: string;
        candidateDigest: string;
        baseRevision: string;
        diff: string;
      }
  > {
    return this.uow.transaction(async (tx) => {
      const agentRunId = unsafeOpaqueId<AgentRunId>(agentRunIdValue);
      const taskId = unsafeOpaqueId<TaskId>(taskIdValue);
      const agentRun = await tx.agentRuns.getById(agentRunId);
      const task = await tx.tasks.getById(taskId);
      if (!agentRun || !task || agentRun.taskId !== task.id) {
        throw new Error("ACP launch hierarchy no longer matches the canonical AgentRun/Task");
      }
      const factoryRun = await tx.factoryRuns.getById(agentRun.factoryRunId);
      const project = factoryRun ? await tx.projects.getById(factoryRun.projectId) : undefined;
      if (!factoryRun || !project || task.projectId !== project.id) {
        throw new Error("ACP launch cannot resolve the canonical Project repository");
      }
      if (agentRun.role === "coder") {
        const prompt = [
          `Implement the AWP Task: ${task.title}`,
          `Project: ${project.name}`,
          "Work only in the provided repository workspace.",
          "Make the smallest complete change that satisfies the task and preserve existing architecture.",
          "Run focused verification when practical.",
          "Do not commit, push, publish, merge, deploy, or request reusable credentials.",
          "Leave the final source changes in the working tree for AWP to collect and review.",
        ].join("\n");
        return { repositoryUrl: project.repositoryUrl, prompt, role: "coder" as const };
      }

      const changeSets = await tx.changeSets.listByProject(project.id);
      const reviews = await tx.reviews.listByChangeSetIds(
        changeSets.map((changeSet) => changeSet.id),
      );
      const review = reviews.find((candidate) => candidate.reviewerAgentRunId === agentRun.id);
      const changeSet = review
        ? changeSets.find((candidate) => candidate.id === review.changeSetId)
        : undefined;
      if (
        !review ||
        !changeSet ||
        review.candidateDigest !== changeSet.candidateDigest ||
        changeSet.status === "merged"
      ) {
        throw new Error("Reviewer AgentRun is not bound to a current exact ChangeSet candidate");
      }
      const evidence = await tx.verificationEvidence.listByChangeSetIds([changeSet.id]);
      const evidenceSummary = evidence
        .filter((item) => item.candidateDigest === changeSet.candidateDigest)
        .map(
          (item) =>
            `- ${item.required ? "required" : "optional"} ${item.source}/${item.name}: ${item.state} @ ${item.observedAt}`,
        );
      const prompt = [
        `Independently review the exact published candidate for AWP Task: ${task.title}`,
        `Project: ${project.name}`,
        `ChangeSet: ${changeSet.id}`,
        `Candidate tree: ${changeSet.candidateDigest}`,
        `Changed paths: ${changeSet.candidateManifest.changedPaths.join(", ")}`,
        "The workspace is already materialized to the exact candidate. Treat it as read-only.",
        "Inspect correctness, architecture, security, tests/evidence, maintainability, and spec conformance.",
        "Do not edit files, commit, push, publish, merge, deploy, or request reusable credentials.",
        "Consume existing verification evidence; do not rerun valid checks without a concrete reason.",
        evidenceSummary.length > 0
          ? `Existing verification evidence:\n${evidenceSummary.join("\n")}`
          : "Existing verification evidence: none",
        'Return only one JSON object: {"disposition":"approved|changes-requested|blocked","findings":[{"severity":"blocking|warning|recommendation|info","summary":"..."}]}.',
        "Use approved only when no blocking correction is required. A non-approved disposition must include at least one finding.",
      ].join("\n");
      return {
        repositoryUrl: project.repositoryUrl,
        prompt,
        role: "reviewer" as const,
        reviewId: review.id,
        candidateDigest: changeSet.candidateDigest,
        baseRevision: changeSet.baseIdentity,
        diff: changeSet.diff,
      };
    });
  }

  private async waitForRunner(workspaceId: string): Promise<void> {
    const deadline = Date.now() + this.readyTimeoutMs;
    let lastError: unknown;
    while (Date.now() < deadline) {
      try {
        const response = await this.runnerRequest(workspaceId, { action: "capabilities" }, false);
        if (response !== null) return;
      } catch (error) {
        lastError = error;
      }
      await new Promise((resolve) => setTimeout(resolve, 500));
    }
    throw new Error(
      `Timed out waiting for ACP workspace runner${lastError instanceof Error ? `: ${lastError.message}` : ""}`,
    );
  }

  private async ensureRepository(
    workspaceId: string,
    repositoryUrl: string,
    expectedBaseRevision?: string,
  ): Promise<string> {
    const pod = resourceBaseName(unsafeOpaqueId<WorkspaceId>(workspaceId));
    const exists = await this.kubectlExec(pod, ["test", "-d", "/workspace/repo/.git"], true);
    if (exists.exitCode !== 0) {
      await this.kubectlExec(pod, ["mkdir", "-p", "/workspace/repo"]);
      if (
        this.options.sourceRepositoryPath &&
        this.options.selfRepositoryUrl &&
        normalizeRepositoryUrl(repositoryUrl) ===
          normalizeRepositoryUrl(this.options.selfRepositoryUrl)
      ) {
        await this.seedSourceRepository(pod, repositoryUrl);
      } else {
        await this.kubectlExec(pod, [
          "git",
          "clone",
          "--no-tags",
          "--",
          repositoryUrl,
          "/workspace/repo",
        ]);
      }
    }
    if (expectedBaseRevision !== undefined) {
      let baseExists = await this.kubectlExec(
        pod,
        ["git", "-C", "/workspace/repo", "cat-file", "-e", `${expectedBaseRevision}^{commit}`],
        true,
      );
      if (baseExists.exitCode !== 0) {
        await this.kubectlExec(pod, [
          "git",
          "-C",
          "/workspace/repo",
          "fetch",
          "--no-tags",
          "origin",
        ]);
        baseExists = await this.kubectlExec(
          pod,
          ["git", "-C", "/workspace/repo", "cat-file", "-e", `${expectedBaseRevision}^{commit}`],
          true,
        );
      }
      if (baseExists.exitCode !== 0) {
        throw new Error("Reviewer workspace cannot resolve the immutable ChangeSet base revision");
      }
      await this.kubectlExec(pod, [
        "git",
        "-C",
        "/workspace/repo",
        "reset",
        "--hard",
        expectedBaseRevision,
      ]);
      await this.kubectlExec(pod, ["git", "-C", "/workspace/repo", "clean", "-ffd"]);
    }
    const revision = await this.kubectlExec(pod, [
      "git",
      "-C",
      "/workspace/repo",
      "rev-parse",
      "HEAD",
    ]);
    const value = revision.stdout.trim();
    if (!/^[0-9a-f]{40,64}$/i.test(value)) {
      throw new Error("Workspace repository did not expose a valid immutable base revision");
    }
    if (expectedBaseRevision !== undefined && value !== expectedBaseRevision) {
      throw new Error("Reviewer workspace base revision does not match the ChangeSet base");
    }
    return value;
  }

  private async materializeReviewCandidate(
    workspaceId: string,
    diff: string,
    candidateDigest: string,
  ): Promise<void> {
    const pod = resourceBaseName(unsafeOpaqueId<WorkspaceId>(workspaceId));
    await this.writeTextToPod(pod, "/workspace/.awp/review.patch", diff);
    await this.kubectlExec(pod, [
      "git",
      "-C",
      "/workspace/repo",
      "apply",
      "--index",
      "--binary",
      "/workspace/.awp/review.patch",
    ]);
    const tree = await this.kubectlExec(pod, ["git", "-C", "/workspace/repo", "write-tree"]);
    if (tree.stdout.trim() !== candidateDigest) {
      throw new Error(
        "Reviewer workspace candidate tree does not match the immutable ChangeSet digest",
      );
    }
    await this.kubectlExec(pod, ["rm", "-f", "/workspace/.awp/review.patch"]);
  }

  private async seedSourceRepository(pod: string, repositoryUrl: string): Promise<void> {
    const source = this.options.sourceRepositoryPath;
    if (!source) throw new Error("Source repository seed path is not configured");
    await access(source);
    const status = await execFileAsync("git", ["-C", source, "status", "--porcelain"], {
      maxBuffer: 4 * 1024 * 1024,
    });
    if (status.stdout.trim()) {
      throw new Error("Source repository snapshot is dirty; refusing to seed an Agent workspace");
    }
    await this.pipeTarToPod(source, pod);
    await this.kubectlExec(pod, [
      "git",
      "-C",
      "/workspace/repo",
      "remote",
      "set-url",
      "origin",
      repositoryUrl,
    ]);
  }

  private async writeTextToPod(pod: string, path: string, content: string): Promise<void> {
    const kubectl = spawn(
      this.kubectl,
      ["-n", this.options.namespace, "exec", "-i", pod, "--", "sh", "-c", `cat > ${path}`],
      { stdio: ["pipe", "pipe", "pipe"] },
    );
    const errors: Buffer[] = [];
    kubectl.stderr.on("data", (chunk: Buffer) => errors.push(chunk));
    kubectl.stdin.end(content);
    const code = await new Promise<number | null>((resolve, reject) => {
      kubectl.once("error", reject);
      kubectl.once("close", resolve);
    });
    if (code !== 0) {
      throw new Error(
        `Failed to materialize reviewer candidate patch (kubectl=${code}): ${Buffer.concat(errors).toString("utf8").slice(-2000)}`,
      );
    }
  }

  private async pipeTarToPod(source: string, pod: string): Promise<void> {
    const tar = spawn("tar", ["-C", source, "-cf", "-", "."], {
      stdio: ["ignore", "pipe", "pipe"],
    });
    const kubectl = spawn(
      this.kubectl,
      [
        "-n",
        this.options.namespace,
        "exec",
        "-i",
        pod,
        "--",
        "tar",
        "-C",
        "/workspace/repo",
        "-xf",
        "-",
      ],
      { stdio: ["pipe", "pipe", "pipe"] },
    );
    tar.stdout.pipe(kubectl.stdin);
    const tarError: Buffer[] = [];
    const kubectlError: Buffer[] = [];
    tar.stderr.on("data", (chunk: Buffer) => tarError.push(chunk));
    kubectl.stderr.on("data", (chunk: Buffer) => kubectlError.push(chunk));
    const [tarCode, kubectlCode] = await Promise.all([
      new Promise<number | null>((resolve, reject) => {
        tar.once("error", reject);
        tar.once("close", resolve);
      }),
      new Promise<number | null>((resolve, reject) => {
        kubectl.once("error", reject);
        kubectl.once("close", resolve);
      }),
    ]);
    if (tarCode !== 0 || kubectlCode !== 0) {
      throw new Error(
        `Failed to seed trusted repository snapshot (tar=${tarCode}, kubectl=${kubectlCode}): ${Buffer.concat(
          [...tarError, ...kubectlError],
        )
          .toString("utf8")
          .slice(-2000)}`,
      );
    }
  }

  private async runnerRequest(
    workspaceId: string,
    request: Readonly<Record<string, unknown>>,
    requireSuccess = true,
  ): Promise<unknown> {
    const pod = resourceBaseName(unsafeOpaqueId<WorkspaceId>(workspaceId));
    const encoded = Buffer.from(JSON.stringify(request)).toString("base64url");
    const result = await this.kubectlExec(
      pod,
      ["node", RUNNER_PATH, "request", encoded],
      !requireSuccess,
    );
    if (result.exitCode !== 0) {
      if (!requireSuccess) throw new Error(result.stderr.trim() || "ACP runner unavailable");
      throw Object.assign(new Error(result.stderr.trim() || "ACP runner request failed"), {
        status: 503,
      });
    }
    const line = result.stdout
      .split("\n")
      .map((value) => value.trim())
      .filter(Boolean)
      .at(-1);
    if (line === undefined) throw new Error("ACP runner returned an empty response");
    return JSON.parse(line) as unknown;
  }

  private async kubectlExec(
    pod: string,
    command: readonly string[],
    allowFailure = false,
  ): Promise<{ exitCode: number; stdout: string; stderr: string }> {
    try {
      const result = await execFileAsync(
        this.kubectl,
        ["-n", this.options.namespace, "exec", pod, "--", ...command],
        { maxBuffer: 64 * 1024 * 1024 },
      );
      return { exitCode: 0, stdout: result.stdout, stderr: result.stderr };
    } catch (error) {
      const typed = error as NodeJS.ErrnoException & {
        stdout?: string;
        stderr?: string;
        code?: number;
      };
      const output = {
        exitCode: typeof typed.code === "number" ? typed.code : 1,
        stdout: typed.stdout ?? "",
        stderr: typed.stderr ?? typed.message,
      };
      if (allowFailure) return output;
      throw Object.assign(new Error(output.stderr.trim() || "kubectl exec failed"), {
        status: 503,
      });
    }
  }

  private nativeSessionIdentity(nativeSessionId: string): { workspaceId: string; key: string } {
    const separator = nativeSessionId.lastIndexOf("/");
    const workspaceId = separator > 0 ? nativeSessionId.slice(0, separator) : "";
    const key = separator > 0 ? nativeSessionId.slice(separator + 1) : "";
    if (!workspaceId || !key || !/^[0-9a-f]{64}$/i.test(key)) {
      throw Object.assign(new Error("Invalid ACP native session identity"), { status: 400 });
    }
    return { workspaceId, key };
  }

  private session(value: unknown): RunnerEnvelopeSession {
    if (!value || typeof value !== "object" || Array.isArray(value)) {
      throw new Error("ACP runner returned an invalid session payload");
    }
    const raw = value as Record<string, unknown>;
    const required = [
      "id",
      "attemptId",
      "agentRunId",
      "taskId",
      "workspaceId",
      "providerId",
      "accountId",
      "model",
      "status",
      "protocolVersion",
      "agentName",
    ] as const;
    for (const key of required) {
      if (typeof raw[key] !== "string" || !String(raw[key]).trim()) {
        throw new Error(`ACP runner session is missing ${key}`);
      }
    }
    if (
      !Array.isArray(raw.capabilities) ||
      raw.capabilities.some((item) => typeof item !== "string")
    ) {
      throw new Error("ACP runner session capabilities are invalid");
    }
    const statuses = new Set([
      "starting",
      "running",
      "waiting",
      "checkpointing",
      "completed",
      "failed",
      "cancelling",
      "cancelled",
    ]);
    if (!statuses.has(String(raw.status))) {
      throw new Error("ACP runner session status is invalid");
    }
    const failureKinds = new Set(["protocol", "agent", "permission", "cancelled", "transport"]);
    if (raw.failureKind !== undefined && !failureKinds.has(String(raw.failureKind))) {
      throw new Error("ACP runner failure kind is invalid");
    }
    const workspaceId = String(raw.workspaceId);
    const failureKind =
      raw.failureKind === undefined
        ? undefined
        : (String(raw.failureKind) as NonNullable<RunnerEnvelopeSession["failureKind"]>);
    return {
      id: `${workspaceId}/${String(raw.id)}`,
      attemptId: String(raw.attemptId),
      agentRunId: String(raw.agentRunId),
      taskId: String(raw.taskId),
      workspaceId,
      providerId: String(raw.providerId),
      accountId: String(raw.accountId),
      model: String(raw.model),
      status: String(raw.status) as RunnerEnvelopeSession["status"],
      protocolVersion: String(raw.protocolVersion),
      agentName: String(raw.agentName),
      capabilities: raw.capabilities as string[],
      ...(failureKind === undefined ? {} : { failureKind }),
      ...(raw.recoverable === true ? { recoverable: true } : {}),
    };
  }
}