fix: add aipod imageRef work-ready runner reuse

2026-06-11 01:21:56 +08:00
parent 59272f8edb
commit 5a6e5a4bbd
22 changed files with 598 additions and 32 deletions
@@ -7,6 +7,8 @@ import type { ExecutionPolicy, JsonRecord } from "../common/types.js";
 import { stableHash, validateEnvName } from "../common/validation.js";
 import { renderRunnerJobManifest } from "../runner/k8s-job.js";
 import type { RunnerSessionPvcOptions, RunnerTransientEnv } from "../runner/k8s-job.js";
+import { staticWorkReadyCapabilitySummary } from "../common/work-ready.js";
+import { resolveRunnerEnvImage } from "../common/env-image-ref.js";

 const reusableCredentialEnvNames = new Set([
  "AUTH_PASSWORD",
@@ -36,6 +38,8 @@ export interface RunnerJobDefaults {
  managerUrl: string;
  image: string;
  sourceCommit: string;
+  envIdentity?: string;
+  artifactCatalogFile?: string;
  serviceAccountName?: string;
  kubectlCommand?: string;
  unideskSshEndpointEnv?: JsonRecord;
@@ -51,6 +55,7 @@ export interface CreateRunnerJobInput extends JsonRecord {
  sourceCommit?: string;
  serviceAccountName?: string;
  idempotencyKey?: string;
+  imageRef?: JsonRecord;
  transientEnv?: JsonRecord[];
 }

@@ -61,8 +66,14 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
  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 envImage = await resolveRunnerEnvImage({
+    ...(options.input.imageRef !== undefined ? { imageRef: options.input.imageRef } : {}),
+    ...(optionalString(options.input.image) ? { explicitImage: optionalString(options.input.image) as string } : {}),
+    defaultImage: options.defaults.image,
+    ...(options.defaults.envIdentity ? { envIdentity: options.defaults.envIdentity } : {}),
+    ...(options.defaults.artifactCatalogFile ? { artifactCatalogFile: options.defaults.artifactCatalogFile } : {}),
+  });
+  const image = envImage.image;
  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;
@@ -79,6 +90,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
    namespace,
    managerUrl,
    sourceCommit,
+    envImage,
    serviceAccountName: serviceAccountName ?? null,
    attemptId: optionalString(options.input.attemptId) ?? null,
    runnerId: optionalString(options.input.runnerId) ?? null,
@@ -144,6 +156,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
    runnerId: render.runnerId,
    namespace: render.namespace,
    jobName: render.jobName,
+    image,
    jobIdentity: {
      kind: "Job",
      namespace: render.namespace,
@@ -158,14 +171,17 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
      runnerId: render.runnerId,
      backendProfile: run.backendProfile,
      managerUrl,
+      image,
      sourceCommit,
      placement: "kubernetes-job",
      logPath: `kubectl -n ${render.namespace} logs job/${render.jobName}`,
    },
+    envImage,
    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 })),
    toolCredentials: summarizeToolCredentials(render.toolCredentials, render.namespace),
    transientEnv: summarizeTransientEnv(transientEnv),
    transientEnvSecret: transientEnvSecretResponse,
+    workReady: staticWorkReadyCapabilitySummary(),
    retention: {
      ttlSecondsAfterFinished: render.ttlSecondsAfterFinished,
    },
@@ -208,6 +224,8 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
    idempotencyKey: idempotencyKey ? "present" : null,
    transientEnv: summarizeTransientEnv(transientEnv),
    transientEnvSecret: transientEnvSecretResponse,
+    envImage,
+    workReady: staticWorkReadyCapabilitySummary(),
    toolCredentials: summarizeToolCredentials(render.toolCredentials, render.namespace),
    sessionRef: summarizeSessionRef(run.sessionRef ?? null),
    resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null),
@@ -49,6 +49,8 @@ export async function dispatchQueueTask(options: DispatchQueueTaskOptions): Prom
    run,
    command,
    runnerJob,
+    envImage: jsonRecordOrNull(runnerJob.envImage),
+    workReady: jsonRecordOrNull(runnerJob.workReady),
    latestAttempt,
    pollCommands: {
      queue: `./scripts/agentrun queue show ${task.id}`,
@@ -127,6 +129,8 @@ function buildRunnerJobInput(task: QueueTaskRecord, commandId: string, input: Js
  copyString("runnerId", "runnerId");
  copyString("sourceCommit", "sourceCommit");
  copyString("serviceAccountName", "serviceAccountName");
+  const imageRef = jsonRecordOrNull(input.imageRef) ?? jsonRecordOrNull(task.metadata.aipodImageRef);
+  if (imageRef) jobInput.imageRef = imageRef;
  if (input.transientEnv !== undefined) {
    if (!Array.isArray(input.transientEnv)) throw new AgentRunError("schema-invalid", "transientEnv must be an array", { httpStatus: 400 });
    jobInput.transientEnv = input.transientEnv.map((item, index) => asRecord(item, `transientEnv[${index}]`));
@@ -140,6 +144,11 @@ function stringFrom(record: JsonRecord, key: string): string {
  return value;
 }

+function jsonRecordOrNull(value: unknown): JsonRecord | null {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
+  return value as JsonRecord;
+}
+
 function optionalString(value: unknown): string | undefined {
  return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
 }
@@ -6,6 +6,7 @@ export function runnerJobStatusSummary(job: RunnerJobRecord, events: RunEvent[]
  const jobIdentity = recordAt(job.result, "jobIdentity");
  const kubernetes = recordAt(job.result, "kubernetes");
  const retention = recordAt(job.result, "retention");
+  const envImage = recordAt(job.result, "envImage");
  const terminalStatus = terminalEvent?.payload.terminalStatus;
  return {
    id: job.id,
@@ -17,6 +18,7 @@ export function runnerJobStatusSummary(job: RunnerJobRecord, events: RunEvent[]
    jobName: job.jobName,
    managerUrl: job.managerUrl,
    image: job.image,
+    envImage,
    sourceCommit: job.sourceCommit,
    serviceAccountName: job.serviceAccountName,
    phase: terminalStatus ? `terminal:${terminalStatus}` : kubernetes.created === true ? "created" : "recorded",
@@ -17,6 +17,7 @@ import type { SessionPvcOptions } from "./session-pvc.js";
 import { getProviderProfileConfig, getProviderProfileValidation, listBackendCapabilities, listProviderProfiles, removeProviderProfile, setProviderProfileConfig, setProviderProfileCredential, showProviderProfile, validateProviderProfile } from "./provider-profiles.js";
 import { listToolCredentials, setGithubSshToolCredential, showToolCredential } from "./tool-credentials.js";
 import { aipodSpecFromInput, applyAipodSpec, deleteAipodSpec, listAipodSpecs, renderAipodSpecByName, showAipodSpec } from "../common/aipod-specs.js";
+import { staticWorkReadyCapabilitySummary } from "../common/work-ready.js";

 function pvcOptions(defaults: { kubectlCommand?: string } | undefined): SessionPvcOptions {
  return defaults?.kubectlCommand ? { kubectlCommand: defaults.kubectlCommand } : {};
@@ -42,6 +43,8 @@ export interface ManagerServerOptions {
    namespace?: string;
    managerUrl?: string;
    image?: string;
+    envIdentity?: string;
+    artifactCatalogFile?: string;
    serviceAccountName?: string;
    kubectlCommand?: string;
    unideskSshEndpointEnv?: JsonRecord;
@@ -211,7 +214,7 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
  if (method === "GET" && (path === "/health" || path === "/health/live" || path === "/health/readiness")) {
    const database = await store.health();
    const ready = path === "/health/live" ? true : database.ready;
-    return { serviceId: "agentrun-mgr", live: true, ready, database, sourceCommit, secretRefs: { databaseUrl: database.adapter === "postgres" ? "redacted" : "not-used", valuesPrinted: false } };
+    return { serviceId: "agentrun-mgr", live: true, ready, database, sourceCommit, runnerWorkReady: staticWorkReadyCapabilitySummary(), secretRefs: { databaseUrl: database.adapter === "postgres" ? "redacted" : "not-used", valuesPrinted: false } };
  }
  if (method === "GET" && path === "/api/v1/backends") return await listBackendCapabilities(providerProfileDefaults) as JsonValue;
  if (method === "GET" && path === "/api/v1/tool-credentials") return await listToolCredentials(toolCredentialDefaults) as JsonValue;
@@ -411,6 +414,8 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
        managerUrl: runnerJobDefaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL ?? `http://agentrun-mgr.${namespace}.svc.cluster.local:8080`,
        image: runnerJobDefaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE ?? "",
        sourceCommit,
+        ...optionalStringRecord("envIdentity", runnerJobDefaults?.envIdentity ?? process.env.AGENTRUN_ENV_IDENTITY),
+        ...optionalStringRecord("artifactCatalogFile", runnerJobDefaults?.artifactCatalogFile ?? process.env.AGENTRUN_ARTIFACT_CATALOG_FILE),
        serviceAccountName: runnerJobDefaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT ?? "agentrun-v01-runner",
        ...(runnerJobDefaults?.kubectlCommand ? { kubectlCommand: runnerJobDefaults.kubectlCommand } : {}),
        ...(runnerJobDefaults?.unideskSshEndpointEnv ? { unideskSshEndpointEnv: runnerJobDefaults.unideskSshEndpointEnv } : {}),
@@ -473,6 +478,8 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
        managerUrl: runnerJobDefaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL ?? `http://agentrun-mgr.${namespace}.svc.cluster.local:8080`,
        image: runnerJobDefaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE ?? "",
        sourceCommit,
+        ...optionalStringRecord("envIdentity", runnerJobDefaults?.envIdentity ?? process.env.AGENTRUN_ENV_IDENTITY),
+        ...optionalStringRecord("artifactCatalogFile", runnerJobDefaults?.artifactCatalogFile ?? process.env.AGENTRUN_ARTIFACT_CATALOG_FILE),
        serviceAccountName: runnerJobDefaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT ?? "agentrun-v01-runner",
        ...(runnerJobDefaults?.kubectlCommand ? { kubectlCommand: runnerJobDefaults.kubectlCommand } : {}),
        ...(runnerJobDefaults?.unideskSshEndpointEnv ? { unideskSshEndpointEnv: runnerJobDefaults.unideskSshEndpointEnv } : {}),
@@ -597,6 +604,10 @@ function stringField(record: JsonRecord, key: string): string {
  return value.trim();
 }

+function optionalStringRecord(key: string, value: unknown): JsonRecord {
+  return typeof value === "string" && value.trim().length > 0 ? { [key]: value.trim() } : {};
+}
+
 function normalizeError(error: unknown): AgentRunError {
  if (error instanceof AgentRunError) return error;
  return new AgentRunError("infra-failed", error instanceof Error ? error.message : String(error), { httpStatus: 500 });