pikasTech-agentrun/src/mgr/kubernetes-runner-job.ts

import { spawn } from "node:child_process";
import { AgentRunError } from "../common/errors.js";
import { redactJson, redactText } from "../common/redaction.js";
import type { AgentRunStore } from "./store.js";
import type { JsonRecord } from "../common/types.js";
import { renderRunnerJobManifest } from "../runner/k8s-job.js";

export interface RunnerJobDefaults {
  namespace: string;
  managerUrl: string;
  image: string;
  sourceCommit: string;
  serviceAccountName?: string;
  kubectlCommand?: string;
}

export interface CreateRunnerJobInput extends JsonRecord {
  commandId: string;
  managerUrl?: string;
  image?: string;
  namespace?: string;
  attemptId?: string;
  runnerId?: string;
  sourceCommit?: string;
  serviceAccountName?: string;
}

export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults }): Promise<JsonRecord> {
  const commandId = stringField(options.input, "commandId");
  const run = await options.store.getRun(options.runId);
  const command = await options.store.getCommand(commandId);
  if (command.runId !== run.id) throw new AgentRunError("schema-invalid", `command ${commandId} does not belong to run ${run.id}`, { httpStatus: 400 });
  if (command.type !== "turn") throw new AgentRunError("schema-invalid", `command ${commandId} is not a turn command`, { httpStatus: 400 });

  const image = optionalString(options.input.image) ?? options.defaults.image;
  if (!image) throw new AgentRunError("schema-invalid", "runner job image is required; set --image or AGENTRUN_RUNNER_IMAGE", { httpStatus: 400 });
  const namespace = optionalString(options.input.namespace) ?? options.defaults.namespace;
  const managerUrl = optionalString(options.input.managerUrl) ?? options.defaults.managerUrl;
  const sourceCommit = optionalString(options.input.sourceCommit) ?? options.defaults.sourceCommit;
  const serviceAccountName = optionalString(options.input.serviceAccountName) ?? options.defaults.serviceAccountName;

  const renderOptions = {
    run,
    commandId,
    managerUrl,
    image,
    namespace,
    sourceCommit,
    ...(serviceAccountName ? { serviceAccountName } : {}),
  };
  const attemptId = optionalString(options.input.attemptId);
  const runnerId = optionalString(options.input.runnerId);
  const render = renderRunnerJobManifest({ ...renderOptions, ...(attemptId ? { attemptId } : {}), ...(runnerId ? { runnerId } : {}) });
  const created = await kubectlCreate(render.manifest, options.defaults.kubectlCommand ?? "kubectl");
  return {
    action: "create-kubernetes-job",
    mutation: true,
    jobIdentity: {
      kind: "Job",
      namespace: render.namespace,
      name: render.jobName,
      serviceAccountName: render.serviceAccountName,
      uid: objectPath(created, ["metadata", "uid"]),
    },
    runner: {
      runId: run.id,
      commandId,
      attemptId: render.attemptId,
      runnerId: render.runnerId,
      backendProfile: run.backendProfile,
      managerUrl,
      sourceCommit,
      placement: "kubernetes-job",
      logPath: `kubectl -n ${render.namespace} logs job/${render.jobName}`,
    },
    secretRefs: render.secretRefs.map((item) => ({ profile: item.profile, name: item.secretRef.name, namespace: item.secretRef.namespace ?? render.namespace, keys: item.secretRef.keys ?? [], mountPath: item.runtimeMountPath, projectionPath: item.projectionMountPath, writableCopy: true, valuesPrinted: false })),
    retention: {
      ttlSecondsAfterFinished: render.ttlSecondsAfterFinished,
    },
    pollCommands: {
      run: `./scripts/agentrun runs show ${run.id} --manager-url ${managerUrl}`,
      command: `./scripts/agentrun commands show ${commandId} --run-id ${run.id} --manager-url ${managerUrl}`,
      events: `./scripts/agentrun runs events ${run.id} --manager-url ${managerUrl} --after-seq 0 --limit 100`,
    },
    warnings: render.warnings,
    kubernetes: {
      created: true,
      valuesPrinted: false,
      apiVersion: objectPath(created, ["apiVersion"]),
      kind: objectPath(created, ["kind"]),
      resourceVersion: objectPath(created, ["metadata", "resourceVersion"]),
    },
  };
}

async function kubectlCreate(manifest: JsonRecord, kubectlCommand: string): Promise<JsonRecord> {
  const child = spawn(kubectlCommand, ["create", "-f", "-", "-o", "json"], { stdio: ["pipe", "pipe", "pipe"] });
  let stdout = "";
  let stderr = "";
  child.stdout.setEncoding("utf8");
  child.stderr.setEncoding("utf8");
  child.stdout.on("data", (chunk) => { stdout += String(chunk); });
  child.stderr.on("data", (chunk) => { stderr += String(chunk); });
  child.stdin.end(`${JSON.stringify(manifest)}\n`);
  const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
    child.on("error", reject);
    child.on("close", (code, signal) => resolve({ code, signal }));
  }).catch((error: unknown) => {
    throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
  });
  if (result.code !== 0) {
    throw new AgentRunError("infra-failed", `kubectl create runner job failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(stderr.slice(-4000)), stdout: redactText(stdout.slice(-2000)), signal: result.signal }) });
  }
  try {
    const parsed = JSON.parse(stdout) as unknown;
    if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) return redactJson(parsed as JsonRecord);
  } catch (error) {
    throw new AgentRunError("infra-failed", `kubectl returned invalid JSON: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { stdoutPreview: redactText(stdout.slice(0, 1000)) } });
  }
  throw new AgentRunError("infra-failed", "kubectl returned non-object JSON", { httpStatus: 502 });
}

function stringField(record: JsonRecord, key: string): string {
  const value = record[key];
  if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${key} is required`, { httpStatus: 400 });
  return value.trim();
}

function optionalString(value: unknown): string | undefined {
  return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
}

function objectPath(value: unknown, path: string[]): string | null {
  let current: unknown = value;
  for (const key of path) {
    if (typeof current !== "object" || current === null || Array.isArray(current)) return null;
    current = (current as Record<string, unknown>)[key];
  }
  return typeof current === "string" ? current : null;
}