Merge pull request #382 from pikasTech/fix/runner-startup-retry-l2

fix: 暴露有限 runner 启动恢复事件

src/mgr/runner-job-status.ts

@@ -9,6 +9,7 @@ export function runnerJobStatusSummary(job: RunnerJobRecord, events: RunEvent[]
   const retention = recordAt(job.result, "retention");
   const envImage = recordAt(job.result, "envImage");
   const reconcilerObservation = recordAt(job.result, "observation");
+  const startupRecovery = recordAt(job.result, "startupRecovery");
   const terminalStatus = observation.terminalStatus;
   return {
     id: job.id,
@@ -35,6 +36,7 @@ export function runnerJobStatusSummary(job: RunnerJobRecord, events: RunEvent[]
     terminalReportState: typeof observation.terminalReportState === "string" ? observation.terminalReportState : null,
     runReportState: typeof observation.runReportState === "string" ? observation.runReportState : null,
     reconcilerObservation,
+    startupRecovery,
     jobIdentity,
     podIdentity: recordAt(job.result, "podIdentity"),
     logPath: typeof runner.logPath === "string" ? runner.logPath : null,

src/mgr/runner-reconciler.ts

@@ -4,6 +4,7 @@ import { externallyAppendableEventTypes } from "../common/events.js";
 import { redactJson, redactText } from "../common/redaction.js";
 import type { BackendEvent, CommandRecord, EventType, FailureKind, JsonRecord, JsonValue, RunnerJobRecord, TerminalStatus } from "../common/types.js";
 import { emitAgentRunOtelSpan } from "../common/otel-trace.js";
+import { agentRunBusinessTraceId } from "../common/otel-trace.js";
 import { nowIso, stableHash } from "../common/validation.js";
 import type { AgentRunStore } from "./store.js";
 import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
@@ -17,6 +18,16 @@ export interface RunnerReconcilerOptions {
   intervalMs?: number;
   nextReconcileAt?: string | null;
   retention?: RunnerRetentionOptions;
+  startupRecovery?: RunnerStartupRecoveryOptions;
+}
+
+export interface RunnerStartupRecoveryOptions {
+  maxAttempts: number;
+  initialBackoffMs: number;
+  maxBackoffMs: number;
+  multiplier: number;
+  totalDeadlineMs: number;
+  jitterRatio: number;
 }
 
 export interface RunnerReconcilerLoopOptions extends RunnerReconcilerOptions {
@@ -82,12 +93,17 @@ export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): P
   const items: JsonRecord[] = [];
   let observeFailedCount = 0;
   let runClosureCount = 0;
+  let startupRecoveryTransitionCount = 0;
 
   for (const job of jobs) {
     let observation = await observeRunnerJob(job, { ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) });
     const terminalOutbox = await recoverTerminalOutboxIfNeeded(input.store, job, observation, { ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) });
     if (terminalOutbox) observation = mergeTerminalOutboxObservation(observation, terminalOutbox);
-    await input.store.updateRunnerJobResult(job.id, { observation });
+    const startupRecovery = input.startupRecovery
+      ? await reconcileRunnerStartupRecovery(input.store, job, observation, input.startupRecovery)
+      : { enabled: false, state: "disabled", valuesPrinted: false };
+    if (startupRecovery.transitioned === true) startupRecoveryTransitionCount++;
+    await input.store.updateRunnerJobResult(job.id, { observation, startupRecovery });
     if (stringValue(observation.category) === "runner-job-observe-failed") observeFailedCount++;
     const runClosure = await closeOpenRunWhenCommandTerminal(input.store, job, observation);
     if (runClosure.closed === true) runClosureCount++;
@@ -101,6 +117,7 @@ export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): P
       terminalReportState: stringValue(observation.terminalReportState) ?? "unknown",
       runReportState: stringValue(observation.runReportState) ?? "unknown",
       terminalOutboxState: terminalOutboxState(observation),
+      startupRecovery,
       runClosure,
       valuesPrinted: false,
     });
@@ -120,6 +137,7 @@ export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): P
     updatedCount: jobs.length,
     observeFailedCount,
     runClosureCount,
+    startupRecoveryTransitionCount,
     stalePendingRetention,
     items,
     valuesPrinted: false,
@@ -346,7 +364,7 @@ export function startRunnerJobReconciler(input: RunnerReconcilerLoopOptions): ()
     input.onSchedule?.(null);
     let result: JsonRecord | null = null;
     try {
-      result = await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}), ...(input.retention ? { retention: input.retention } : {}) });
+      result = await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}), ...(input.retention ? { retention: input.retention } : {}), ...(input.startupRecovery ? { startupRecovery: input.startupRecovery } : {}) });
     } catch (error) {
       input.onError?.(error);
     } finally {
@@ -467,8 +485,9 @@ function observedRunnerPhase(jobObject: K8sObject | null, pods: K8sObject[]): st
   const podPhases = pods.map((pod) => stringPath(pod, ["status", "phase"])).filter((phase): phase is string => Boolean(phase));
   if ((succeeded ?? 0) > 0 || podPhases.some((phase) => phase === "Succeeded")) return "k8s:succeeded";
   if ((failed ?? 0) > 0 || podPhases.some((phase) => phase === "Failed")) return "k8s:failed";
-  if ((active ?? 0) > 0 || podPhases.some((phase) => phase === "Running")) return "k8s:running";
+  if (podPhases.some((phase) => phase === "Running")) return "k8s:running";
   if (podPhases.some((phase) => phase === "Pending")) return "k8s:pending";
+  if ((active ?? 0) > 0) return "k8s:pending";
   return "k8s:unknown";
 }
 
@@ -488,6 +507,7 @@ function terminalReportStateForPhase(phase: string): string {
 
 function k8sSummary(jobObject: K8sObject | null, pods: K8sObject[]): JsonRecord {
   const podPhases = pods.map((pod) => stringPath(pod, ["status", "phase"]) ?? "unknown");
+  const waiting = newestContainerWaiting(pods);
   return {
     jobPresent: jobObject !== null,
     jobUid: stringPath(jobObject, ["metadata", "uid"]),
@@ -500,10 +520,301 @@ function k8sSummary(jobObject: K8sObject | null, pods: K8sObject[]): JsonRecord
     podCount: pods.length,
     podPhases,
     newestPodName: newestPodName(pods),
+    waiting,
     valuesPrinted: false,
   };
 }
 
+interface StartupFailure {
+  failureDomain: "infrastructure";
+  component: string;
+  code: string;
+  summary: string;
+  retryable: boolean;
+}
+
+async function reconcileRunnerStartupRecovery(
+  store: AgentRunStore,
+  job: RunnerJobRecord,
+  observation: JsonRecord,
+  policy: RunnerStartupRecoveryOptions,
+): Promise<JsonRecord> {
+  const previous = recordValue(job.result.startupRecovery) ?? {};
+  const previousState = stringValue(previous.state);
+  const failure = startupFailureFromObservation(observation);
+  const observedAt = stringValue(observation.lastK8sObservedAt) ?? nowIso();
+  if (!failure) {
+    if (previousState !== "scheduled" && previousState !== "retrying") {
+      return previousState ? { ...previous, enabled: true, transitioned: false, valuesPrinted: false } : { enabled: true, state: "healthy", transitioned: false, valuesPrinted: false };
+    }
+    const transition = await emitStartupTransition(store, job, {
+      phase: "runner-startup-retry-recovered",
+      state: "recovered",
+      failure: failureFromRecoveryState(previous),
+      attempt: integerValue(previous.attempt) ?? 1,
+      maxAttempts: policy.maxAttempts,
+      firstObservedAt: stringValue(previous.firstObservedAt) ?? observedAt,
+      observedAt,
+      backoffMs: 0,
+      nextRetryAt: null,
+    });
+    return { ...transition.state, enabled: true, transitioned: transition.appended, valuesPrinted: false };
+  }
+
+  const firstObservedAt = stringValue(previous.firstObservedAt) ?? observedAt;
+  const deadlineAt = new Date(Date.parse(firstObservedAt) + policy.totalDeadlineMs).toISOString();
+  const deadlineExceeded = Date.parse(observedAt) >= Date.parse(deadlineAt);
+  const previousFingerprint = stringValue(previous.failureFingerprint);
+  const failureFingerprint = stableHash({ ...failure });
+  const sameFailure = previousFingerprint === failureFingerprint;
+  const previousAttempt = integerValue(previous.attempt) ?? 0;
+  const nextRetryAt = stringValue(previous.nextRetryAt);
+  const retryDue = nextRetryAt !== null && Date.parse(observedAt) >= Date.parse(nextRetryAt);
+
+  if (previousState === "exhausted") {
+    await terminalizeStartupFailure(store, job, failure);
+    return { ...previous, enabled: true, transitioned: false, valuesPrinted: false };
+  }
+  if (sameFailure && previousState === "scheduled" && !retryDue) {
+    return { ...previous, enabled: true, transitioned: false, valuesPrinted: false };
+  }
+
+  let attempt = sameFailure ? Math.max(1, previousAttempt) : 1;
+  if (sameFailure && previousState === "scheduled" && retryDue) {
+    await emitStartupTransition(store, job, {
+      phase: "runner-startup-retry-started",
+      state: "retrying",
+      failure,
+      attempt,
+      maxAttempts: policy.maxAttempts,
+      firstObservedAt,
+      observedAt,
+      backoffMs: 0,
+      nextRetryAt: null,
+    });
+    attempt += 1;
+  }
+
+  const exhausted = !failure.retryable || deadlineExceeded || attempt > policy.maxAttempts;
+  if (exhausted) {
+    if (!sameFailure) {
+      await emitStartupTransition(store, job, {
+        phase: "runner-startup-failure-observed",
+        state: "observed",
+        failure,
+        attempt: Math.min(attempt, policy.maxAttempts),
+        maxAttempts: policy.maxAttempts,
+        firstObservedAt,
+        observedAt,
+        backoffMs: 0,
+        nextRetryAt: null,
+        deadlineAt,
+      });
+    }
+    const transition = await emitStartupTransition(store, job, {
+      phase: "runner-startup-retry-exhausted",
+      state: "exhausted",
+      failure,
+      attempt: Math.min(attempt, policy.maxAttempts),
+      maxAttempts: policy.maxAttempts,
+      firstObservedAt,
+      observedAt,
+      backoffMs: 0,
+      nextRetryAt: null,
+      deadlineAt,
+    });
+    await terminalizeStartupFailure(store, job, failure);
+    return { ...transition.state, enabled: true, transitioned: transition.appended, valuesPrinted: false };
+  }
+
+  const backoffMs = startupBackoffMs(policy, attempt, job.id);
+  const scheduledAt = new Date(Date.parse(observedAt) + backoffMs).toISOString();
+  const observed = await emitStartupTransition(store, job, {
+    phase: "runner-startup-failure-observed",
+    state: "observed",
+    failure,
+    attempt,
+    maxAttempts: policy.maxAttempts,
+    firstObservedAt,
+    observedAt,
+    backoffMs,
+    nextRetryAt: scheduledAt,
+    deadlineAt,
+  });
+  const scheduled = await emitStartupTransition(store, job, {
+    phase: "runner-startup-retry-scheduled",
+    state: "scheduled",
+    failure,
+    attempt,
+    maxAttempts: policy.maxAttempts,
+    firstObservedAt,
+    observedAt,
+    backoffMs,
+    nextRetryAt: scheduledAt,
+    deadlineAt,
+  });
+  return { ...scheduled.state, enabled: true, transitioned: observed.appended || scheduled.appended, valuesPrinted: false };
+}
+
+function startupFailureFromObservation(observation: JsonRecord): StartupFailure | null {
+  const phase = stringValue(observation.observedRunnerPhase);
+  if (phase === "k8s:observe-failed") {
+    return {
+      failureDomain: "infrastructure",
+      component: "kubernetes-api",
+      code: "kubernetes-observation-failed",
+      summary: "Kubernetes runner 状态暂时不可读取",
+      retryable: true,
+    };
+  }
+  if (phase !== "k8s:pending") return null;
+  const waiting = recordValue(recordValue(observation.k8s)?.waiting);
+  const reason = stringValue(waiting?.reason) ?? "";
+  const message = stringValue(waiting?.message) ?? "";
+  const detail = `${reason} ${message}`.toLowerCase();
+  if (/unauthorized|authentication required|pull access denied|denied|forbidden/u.test(detail)) {
+    return { failureDomain: "infrastructure", component: "runner-image-pull", code: "runner-image-auth-denied", summary: "Runner 镜像仓库鉴权被拒绝", retryable: false };
+  }
+  if (/manifest unknown|not found|no such manifest|failed to resolve reference.*not found/u.test(detail)) {
+    return { failureDomain: "infrastructure", component: "runner-image-pull", code: "runner-image-not-found", summary: "Runner 镜像或摘要不存在", retryable: false };
+  }
+  if (/invalidimagename|invalid image|invalid reference format/u.test(detail)) {
+    return { failureDomain: "infrastructure", component: "runner-image-pull", code: "runner-image-invalid", summary: "Runner 镜像地址无效", retryable: false };
+  }
+  if (/createcontainerconfigerror|secret .* not found|configmap .* not found/u.test(detail)) {
+    return { failureDomain: "infrastructure", component: "runner-startup-config", code: "runner-startup-config-missing", summary: "Runner 启动所需配置或 Secret 缺失", retryable: false };
+  }
+  if (/imagepullbackoff|errimagepull|connection refused|i\/o timeout|no such host|temporary failure|tls handshake timeout|service unavailable/u.test(detail)) {
+    return { failureDomain: "infrastructure", component: "runner-image-pull", code: "runner-image-registry-unreachable", summary: "Runner 镜像仓库暂时不可达", retryable: true };
+  }
+  return null;
+}
+
+function newestContainerWaiting(pods: K8sObject[]): JsonRecord | null {
+  const sorted = [...pods].sort((left, right) => (stringPath(right, ["metadata", "creationTimestamp"]) ?? "").localeCompare(stringPath(left, ["metadata", "creationTimestamp"]) ?? ""));
+  for (const pod of sorted) {
+    for (const path of [["status", "initContainerStatuses"], ["status", "containerStatuses"]]) {
+      const statuses = arrayPath(pod, path);
+      for (const status of statuses) {
+        const waiting = recordValue(recordValue(status)?.state)?.waiting;
+        const record = recordValue(waiting);
+        if (!record) continue;
+        return {
+          containerName: stringValue(recordValue(status)?.name) ?? null,
+          reason: stringValue(record.reason),
+          message: stringValue(record.message) ? redactText(stringValue(record.message) as string).slice(0, 1_000) : null,
+          valuesPrinted: false,
+        };
+      }
+    }
+  }
+  return null;
+}
+
+async function emitStartupTransition(
+  store: AgentRunStore,
+  job: RunnerJobRecord,
+  input: {
+    phase: string;
+    state: string;
+    failure: StartupFailure;
+    attempt: number;
+    maxAttempts: number;
+    firstObservedAt: string;
+    observedAt: string;
+    backoffMs: number;
+    nextRetryAt: string | null;
+    deadlineAt?: string;
+  },
+): Promise<{ state: JsonRecord; appended: boolean }> {
+  const run = await store.getRun(job.runId);
+  const command = await store.getCommand(job.commandId);
+  const transitionKey = stableHash({
+    runnerJobId: job.id,
+    phase: input.phase,
+    attempt: input.attempt,
+    code: input.failure.code,
+    nextRetryAt: input.nextRetryAt,
+  });
+  const payload: JsonRecord = {
+    phase: input.phase,
+    retryPhase: input.phase.replace(/^runner-startup-/u, ""),
+    failureDomain: input.failure.failureDomain,
+    component: input.failure.component,
+    code: input.failure.code,
+    summary: input.failure.summary,
+    failureKind: "infra-failed",
+    retryable: input.failure.retryable,
+    attempt: input.attempt,
+    maxAttempts: input.maxAttempts,
+    backoffMs: input.backoffMs,
+    nextRetryAt: input.nextRetryAt,
+    firstObservedAt: input.firstObservedAt,
+    observedAt: input.observedAt,
+    deadlineAt: input.deadlineAt ?? null,
+    runId: job.runId,
+    commandId: job.commandId,
+    runnerJobId: job.id,
+    traceId: agentRunBusinessTraceId(run, command),
+    transitionKey,
+    valuesPrinted: false,
+  };
+  const existing = await store.listEventsForCommand(job.runId, job.commandId, 2_000);
+  const appended = !existing.some((event) => event.payload.transitionKey === transitionKey);
+  if (appended) await store.appendEvent(job.runId, "backend_status", payload);
+  const state: JsonRecord = {
+    state: input.state,
+    failureFingerprint: stableHash({ ...input.failure }),
+    ...payload,
+    transitioned: appended,
+    valuesPrinted: false,
+  };
+  return { state, appended };
+}
+
+async function terminalizeStartupFailure(store: AgentRunStore, job: RunnerJobRecord, failure: StartupFailure): Promise<void> {
+  const command = await store.getCommand(job.commandId);
+  if (!isTerminalCommandState(command.state)) {
+    await store.finishCommand(job.commandId, { terminalStatus: "failed", failureKind: "infra-failed", failureMessage: failure.summary });
+  }
+  const run = await store.getRun(job.runId);
+  if (!isTerminalRunStatus(run.status)) {
+    await store.finishRun(job.runId, { terminalStatus: "failed", failureKind: "infra-failed", failureMessage: failure.summary });
+  }
+}
+
+function failureFromRecoveryState(state: JsonRecord): StartupFailure {
+  return {
+    failureDomain: "infrastructure",
+    component: stringValue(state.component) ?? "runner-startup",
+    code: stringValue(state.code) ?? "runner-startup-recovered",
+    summary: stringValue(state.summary) ?? "Runner 启动故障已恢复",
+    retryable: state.retryable !== false,
+  };
+}
+
+function startupBackoffMs(policy: RunnerStartupRecoveryOptions, attempt: number, seed: string): number {
+  const exponential = Math.min(policy.maxBackoffMs, Math.round(policy.initialBackoffMs * (policy.multiplier ** Math.max(0, attempt - 1))));
+  if (policy.jitterRatio <= 0) return exponential;
+  const hash = Number.parseInt(stableHash({ seed, attempt }).slice(0, 8), 16);
+  const unit = Number.isFinite(hash) ? hash / 0xffffffff : 0.5;
+  const factor = 1 - policy.jitterRatio + (2 * policy.jitterRatio * unit);
+  return Math.max(1, Math.min(policy.maxBackoffMs, Math.round(exponential * factor)));
+}
+
+function integerValue(value: unknown): number | null {
+  return typeof value === "number" && Number.isSafeInteger(value) ? value : null;
+}
+
+function arrayPath(value: unknown, path: string[]): unknown[] {
+  let current: unknown = value;
+  for (const key of path) {
+    if (typeof current !== "object" || current === null || Array.isArray(current)) return [];
+    current = (current as Record<string, unknown>)[key];
+  }
+  return Array.isArray(current) ? current : [];
+}
+
 function jobCondition(jobObject: K8sObject | null): string | null {
   const conditions = jobObject?.status?.conditions;
   if (!Array.isArray(conditions)) return null;

src/mgr/server.ts

@@ -14,7 +14,7 @@ import { dispatchQueueTask, refreshQueueTaskFromCore } from "./queue-dispatch.js
 import { retryQueueTask } from "./queue-retry.js";
 import { buildRunResult } from "./result.js";
 import { runnerJobStatusSummary } from "./runner-job-status.js";
-import { reconcileRunnerJobsOnce, startRunnerJobReconciler } from "./runner-reconciler.js";
+import { reconcileRunnerJobsOnce, startRunnerJobReconciler, type RunnerStartupRecoveryOptions } from "./runner-reconciler.js";
 import { assertRunnerDispatcherAttemptTiming, dispatchRunnerIntentsOnce, startRunnerDispatcher, type RunnerDispatcherOptions } from "./runner-dispatcher.js";
 import { kafkaOutboxRelayStatus, relayKafkaEventOutboxOnce, startKafkaEventOutboxRelay, type KafkaOutboxRelayOptions } from "./kafka-outbox-relay.js";
 import { createSessionPvc, deleteSessionPvc, getSessionPvcSummary, refreshSessionPvcSummary, runSessionStorageGc } from "./session-pvc.js";
@@ -122,7 +122,32 @@ function runnerReconcilerOptionsForRuntime(defaults: ManagerServerOptions["runne
   const intervalMs = defaults?.intervalMs ?? optionalPositiveInteger("AGENTRUN_MANAGER_RECONCILER_INTERVAL_MS", process.env.AGENTRUN_MANAGER_RECONCILER_INTERVAL_MS) ?? 30_000;
   const batchSize = defaults?.batchSize ?? optionalPositiveInteger("AGENTRUN_MANAGER_RECONCILER_BATCH_SIZE", process.env.AGENTRUN_MANAGER_RECONCILER_BATCH_SIZE) ?? 20;
   const kubectlCommand = defaults?.kubectlCommand ?? runnerJobDefaults?.kubectlCommand;
-  return { enabled, namespace, intervalMs, batchSize, nextReconcileAt: null, ...(kubectlCommand ? { kubectlCommand } : {}), ...(runnerJobDefaults.retention ? { retention: runnerJobDefaults.retention } : {}) };
+  const startupRecovery = startupRecoveryOptionsForRuntime(defaults?.startupRecovery);
+  return { enabled, namespace, intervalMs, batchSize, nextReconcileAt: null, ...(kubectlCommand ? { kubectlCommand } : {}), ...(runnerJobDefaults.retention ? { retention: runnerJobDefaults.retention } : {}), ...(startupRecovery ? { startupRecovery } : {}) };
+}
+
+function startupRecoveryOptionsForRuntime(defaults: RunnerStartupRecoveryOptions | undefined): RunnerStartupRecoveryOptions | undefined {
+  if (defaults) return validateStartupRecoveryOptions(defaults);
+  const maxAttempts = optionalPositiveInteger("AGENTRUN_RUNNER_STARTUP_RETRY_MAX_ATTEMPTS", process.env.AGENTRUN_RUNNER_STARTUP_RETRY_MAX_ATTEMPTS);
+  if (maxAttempts === undefined) return undefined;
+  return validateStartupRecoveryOptions({
+    maxAttempts,
+    initialBackoffMs: requiredPositiveIntegerValue("AGENTRUN_RUNNER_STARTUP_RETRY_INITIAL_BACKOFF_MS", process.env.AGENTRUN_RUNNER_STARTUP_RETRY_INITIAL_BACKOFF_MS),
+    maxBackoffMs: requiredPositiveIntegerValue("AGENTRUN_RUNNER_STARTUP_RETRY_MAX_BACKOFF_MS", process.env.AGENTRUN_RUNNER_STARTUP_RETRY_MAX_BACKOFF_MS),
+    multiplier: requiredPositiveNumber("AGENTRUN_RUNNER_STARTUP_RETRY_MULTIPLIER", process.env.AGENTRUN_RUNNER_STARTUP_RETRY_MULTIPLIER),
+    totalDeadlineMs: requiredPositiveIntegerValue("AGENTRUN_RUNNER_STARTUP_RETRY_TOTAL_DEADLINE_MS", process.env.AGENTRUN_RUNNER_STARTUP_RETRY_TOTAL_DEADLINE_MS),
+    jitterRatio: requiredRatio("AGENTRUN_RUNNER_STARTUP_RETRY_JITTER_RATIO", process.env.AGENTRUN_RUNNER_STARTUP_RETRY_JITTER_RATIO),
+  });
+}
+
+function validateStartupRecoveryOptions(options: RunnerStartupRecoveryOptions): RunnerStartupRecoveryOptions {
+  if (options.maxAttempts < 1 || !Number.isSafeInteger(options.maxAttempts)) throw new AgentRunError("schema-invalid", "runner startup retry maxAttempts must be a positive integer", { httpStatus: 500 });
+  if (options.initialBackoffMs < 1 || !Number.isSafeInteger(options.initialBackoffMs)) throw new AgentRunError("schema-invalid", "runner startup retry initialBackoffMs must be a positive integer", { httpStatus: 500 });
+  if (options.maxBackoffMs < options.initialBackoffMs || !Number.isSafeInteger(options.maxBackoffMs)) throw new AgentRunError("schema-invalid", "runner startup retry maxBackoffMs must be an integer greater than or equal to initialBackoffMs", { httpStatus: 500 });
+  if (!Number.isFinite(options.multiplier) || options.multiplier < 1) throw new AgentRunError("schema-invalid", "runner startup retry multiplier must be at least 1", { httpStatus: 500 });
+  if (options.totalDeadlineMs < options.initialBackoffMs || !Number.isSafeInteger(options.totalDeadlineMs)) throw new AgentRunError("schema-invalid", "runner startup retry totalDeadlineMs must cover the initial backoff", { httpStatus: 500 });
+  if (!Number.isFinite(options.jitterRatio) || options.jitterRatio < 0 || options.jitterRatio > 1) throw new AgentRunError("schema-invalid", "runner startup retry jitterRatio must be between 0 and 1", { httpStatus: 500 });
+  return options;
 }
 
 function runnerReconcilerLogSummary(result: JsonRecord): JsonRecord {
@@ -219,7 +244,7 @@ export interface ManagerServerOptions {
   sessionPvcOptions?: { kubectlHandler?: import("./session-pvc.js").KubectlHandler; kubectlCommand?: string; storageClassName?: string; size?: string };
   providerProfileOptions?: { namespace?: string; kubectlCommand?: string };
   toolCredentialOptions?: { namespace?: string; kubectlCommand?: string };
-  runnerReconcilerOptions?: { enabled?: boolean; namespace?: string; kubectlCommand?: string; intervalMs?: number; batchSize?: number };
+  runnerReconcilerOptions?: { enabled?: boolean; namespace?: string; kubectlCommand?: string; intervalMs?: number; batchSize?: number; startupRecovery?: RunnerStartupRecoveryOptions };
   runnerDispatcherOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; maxAttempts?: number; retryBackoffMs?: number; attemptTimeoutMs?: number; owner?: string };
   kafkaOutboxRelayOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; retryBackoffMs?: number; owner?: string; config?: AgentRunKafkaConfig };
   aipodSpecDir?: string;
@@ -234,6 +259,7 @@ interface RunnerReconcilerRuntimeOptions {
   nextReconcileAt: string | null;
   kubectlCommand?: string;
   retention?: RunnerRetentionOptions;
+  startupRecovery?: RunnerStartupRecoveryOptions;
 }
 
 export interface StartedManagerServer {
@@ -264,6 +290,7 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr
     batchSize: runnerReconcilerOptions.batchSize,
     ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}),
     ...(runnerReconcilerOptions.retention ? { retention: runnerReconcilerOptions.retention } : {}),
+    ...(runnerReconcilerOptions.startupRecovery ? { startupRecovery: runnerReconcilerOptions.startupRecovery } : {}),
     onSchedule: (nextReconcileAt) => { runnerReconcilerOptions.nextReconcileAt = nextReconcileAt; },
     onResult: (result) => console.info(JSON.stringify(runnerReconcilerLogSummary(result))),
     onError: (error) => console.warn(JSON.stringify({ code: "runner-reconciler-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })),
@@ -1039,6 +1066,7 @@ async function route({ method, url, body, signal, store, sourceCommit, authSumma
       nextReconcileAt: runnerReconcilerOptions.nextReconcileAt,
       ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}),
       ...(runnerReconcilerOptions.retention ? { retention: runnerReconcilerOptions.retention } : {}),
+      ...(runnerReconcilerOptions.startupRecovery ? { startupRecovery: runnerReconcilerOptions.startupRecovery } : {}),
     }) as JsonValue;
   }
   if (method === "POST" && path === "/api/v1/reconciler/runner-dispatch") {
@@ -1574,6 +1602,24 @@ function requiredPositiveInteger(key: string, value: unknown): number {
   return parsed;
 }
 
+function requiredPositiveIntegerValue(key: string, value: unknown): number {
+  const parsed = optionalPositiveInteger(key, value);
+  if (parsed === undefined) throw new AgentRunError("schema-invalid", `${key} is required when runner startup retry is enabled`, { httpStatus: 500 });
+  return parsed;
+}
+
+function requiredPositiveNumber(key: string, value: unknown): number {
+  const parsed = Number(value);
+  if (!Number.isFinite(parsed) || parsed <= 0) throw new AgentRunError("schema-invalid", `${key} must be a positive number`, { httpStatus: 500 });
+  return parsed;
+}
+
+function requiredRatio(key: string, value: unknown): number {
+  const parsed = Number(value);
+  if (!Number.isFinite(parsed) || parsed < 0 || parsed > 1) throw new AgentRunError("schema-invalid", `${key} must be between 0 and 1`, { httpStatus: 500 });
+  return parsed;
+}
+
 function jsonRecordEnv(key: string, value: unknown): Record<string, string> {
   if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${key} is required when runner retention is enabled`, { httpStatus: 500 });
   const parsed = JSON.parse(value) as unknown;

src/runner/run-once.ts

@@ -6,7 +6,7 @@ import { materializeResourceBundle } from "./resource-bundle.js";
 import type { BackendEvent, BackendProfile, BackendTurnResult, CommandRecord, FailureKind, InitialPromptAssembly, JsonRecord, RunRecord, RunnerRecord, TerminalStatus } from "../common/types.js";
 import { AgentRunError } from "../common/errors.js";
 import { smokeBundledWorkReadyCapabilities, smokeImageWorkReadyCapabilities } from "../common/work-ready.js";
-import { stableHash } from "../common/validation.js";
+import { nowIso, stableHash } from "../common/validation.js";
 import { prepareGithubSshCredentialEnvironment } from "./github-ssh-credential.js";
 
 export interface RunnerOnceOptions extends BackendAdapterOptions {
@@ -332,17 +332,32 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
       return { terminalStatus: "failed", failureKind, failureMessage: message, events: [], ...(result.threadId ? { threadId: result.threadId } : {}), ...(result.turnId ? { turnId: result.turnId } : {}) };
     };
     let result = await runBackendAttempt();
+    let backendFailureFirstObservedAt: string | null = null;
+    let backendLastFailureKind: FailureKind | null = null;
     while (shouldRetryBackendTurn(result, retryPolicy, backendAttempt)) {
+      backendLastFailureKind = result.failureKind;
       const retryDelayMs = backendRetryDelayMs(retryPolicy, backendAttempt);
+      const observedAt = nowIso();
+      backendFailureFirstObservedAt ??= observedAt;
+      const retryNextAt = new Date(Date.parse(observedAt) + retryDelayMs).toISOString();
       const retryPayload = {
-        phase: "runner:backend-retry",
         commandId: command.id,
         attemptId,
         runnerId: runner.id,
+        failureDomain: "upstream",
+        component: "provider-adapter",
+        code: result.failureKind ?? "backend-failed",
+        summary: result.failureMessage ?? `${result.failureKind} retryable backend failure`,
         failureKind: result.failureKind,
         message: result.failureMessage ?? `${result.failureKind} retryable backend failure`,
         retryable: true,
         willRetry: true,
+        attempt: backendAttempt,
+        maxAttempts: retryPolicy.maxAttempts,
+        backoffMs: retryDelayMs,
+        nextRetryAt: retryNextAt,
+        firstObservedAt: backendFailureFirstObservedAt,
+        observedAt,
         retryAttempt: backendAttempt,
         retryNextAttempt: backendAttempt + 1,
         retryMaxAttempts: retryPolicy.maxAttempts,
@@ -350,23 +365,90 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
         retryExhausted: false,
         ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}),
       };
-      await appendBestEffort(api, options.runId, { type: "error", payload: retryPayload });
-      await appendBestEffort(api, options.runId, { type: "backend_status", payload: retryPayload });
-      await runnerLog.write("command.retrying", { runId: options.runId, ...retryPayload, valuesPrinted: false });
+      await appendBestEffort(api, options.runId, { type: "error", payload: { ...retryPayload, phase: "runner-backend-failure-observed", retryPhase: "failureObserved" } });
+      await appendBestEffort(api, options.runId, { type: "backend_status", payload: { ...retryPayload, phase: "runner-backend-retry-scheduled", retryPhase: "retryScheduled" } });
+      await runnerLog.write("command.retrying", { runId: options.runId, ...retryPayload, phase: "runner-backend-retry-scheduled", valuesPrinted: false });
       if (backendSession) {
         const closeEvents = await backendSession.close();
         for (const event of closeEvents) await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
       }
       await sleep(retryDelayMs);
       backendAttempt += 1;
+      await appendBestEffort(api, options.runId, {
+        type: "backend_status",
+        payload: {
+          ...retryPayload,
+          phase: "runner-backend-retry-started",
+          retryPhase: "retryStarted",
+          attempt: backendAttempt,
+          retryAttempt: backendAttempt,
+          backoffMs: 0,
+          retryDelayMs: 0,
+          nextRetryAt: null,
+          observedAt: nowIso(),
+        },
+      });
       result = await runBackendAttempt();
     }
+    if (backendFailureFirstObservedAt && result.terminalStatus === "completed") {
+      await appendBestEffort(api, options.runId, {
+        type: "backend_status",
+        payload: {
+          phase: "runner-backend-retry-recovered",
+          retryPhase: "retryRecovered",
+          commandId: command.id,
+          attemptId,
+          runnerId: runner.id,
+          failureDomain: backendFailureDomain(backendLastFailureKind),
+          component: backendFailureComponent(backendLastFailureKind),
+          code: "provider-recovered",
+          summary: "上游模型服务已恢复",
+          failureKind: null,
+          retryable: true,
+          attempt: backendAttempt,
+          maxAttempts: retryPolicy.maxAttempts,
+          backoffMs: 0,
+          nextRetryAt: null,
+          firstObservedAt: backendFailureFirstObservedAt,
+          observedAt: nowIso(),
+          ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}),
+        },
+      });
+    }
     for (const event of result.events) {
       if (shouldKeepTerminalOutboxEvent(event)) terminalOutboxEvents.push(event);
       else await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
     }
+    if (result.terminalStatus !== "completed" && result.failureKind) {
+      const observedAt = nowIso();
+      backendFailureFirstObservedAt ??= observedAt;
+      terminalOutboxEvents.push({
+        type: "error",
+        payload: {
+          phase: "runner-backend-failure-observed",
+          retryPhase: "failureObserved",
+          commandId: command.id,
+          attemptId,
+          runnerId: runner.id,
+          failureDomain: backendFailureDomain(result.failureKind),
+          component: backendFailureComponent(result.failureKind),
+          code: result.failureKind,
+          summary: result.failureMessage ?? `${result.failureKind} backend failure`,
+          failureKind: result.failureKind,
+          message: result.failureMessage ?? `${result.failureKind} backend failure`,
+          retryable: isRetryableBackendFailure(result.failureKind),
+          attempt: backendAttempt,
+          maxAttempts: retryPolicy.maxAttempts,
+          backoffMs: 0,
+          nextRetryAt: null,
+          firstObservedAt: backendFailureFirstObservedAt,
+          observedAt,
+          ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}),
+        },
+      });
+    }
     if (isRetryableBackendFailure(result.failureKind) && retryPolicy.maxAttempts > 1 && backendAttempt >= retryPolicy.maxAttempts && result.terminalStatus !== "completed") {
-      terminalOutboxEvents.push({ type: "error", payload: { phase: "runner:backend-retry-exhausted", commandId: command.id, attemptId, runnerId: runner.id, failureKind: result.failureKind, message: result.failureMessage ?? `${result.failureKind} retry attempts exhausted`, retryable: true, willRetry: false, retryAttempt: backendAttempt, retryMaxAttempts: retryPolicy.maxAttempts, retryExhausted: true, ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}) } });
+      terminalOutboxEvents.push({ type: "error", payload: { phase: "runner-backend-retry-exhausted", retryPhase: "retryExhausted", commandId: command.id, attemptId, runnerId: runner.id, failureDomain: "upstream", component: "provider-adapter", code: result.failureKind ?? "backend-failed", summary: result.failureMessage ?? `${result.failureKind} retry attempts exhausted`, failureKind: result.failureKind, message: result.failureMessage ?? `${result.failureKind} retry attempts exhausted`, retryable: true, willRetry: false, attempt: backendAttempt, maxAttempts: retryPolicy.maxAttempts, backoffMs: 0, nextRetryAt: null, firstObservedAt: backendFailureFirstObservedAt ?? nowIso(), observedAt: nowIso(), retryAttempt: backendAttempt, retryMaxAttempts: retryPolicy.maxAttempts, retryExhausted: true, ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}) } });
     }
     const report: CommandTerminalReport = { terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: result.failureMessage, ...(result.threadId ? { threadId: result.threadId } : {}), ...(result.turnId ? { turnId: result.turnId } : {}) };
     await reportTerminalCommand(api, {
@@ -411,6 +493,14 @@ function isRetryableBackendFailure(failureKind: FailureKind | null): boolean {
   return failureKind === "backend-timeout" || failureKind === "provider-stream-disconnected" || failureKind === "provider-unavailable" || failureKind === "provider-rate-limited" || failureKind === "provider-http-error";
 }
 
+function backendFailureDomain(failureKind: FailureKind | null): "upstream" | "infrastructure" {
+  return String(failureKind ?? "").startsWith("provider-") || failureKind === "backend-timeout" ? "upstream" : "infrastructure";
+}
+
+function backendFailureComponent(failureKind: FailureKind | null): string {
+  return backendFailureDomain(failureKind) === "upstream" ? "provider-adapter" : "runner-backend";
+}
+
 function backendRetryDelayMs(policy: BackendRetryPolicy, backendAttempt: number): number {
   const exponent = Math.max(0, backendAttempt - 1);
   const delay = policy.initialBackoffMs * (2 ** exponent);

src/selftest/cases/24-runner-startup-retry.ts

@@ -0,0 +1,138 @@
+import assert from "node:assert/strict";
+import { chmod, writeFile } from "node:fs/promises";
+import path from "node:path";
+import type { JsonRecord } from "../../common/types.js";
+import { reconcileRunnerJobsOnce, type RunnerStartupRecoveryOptions } from "../../mgr/runner-reconciler.js";
+import { MemoryAgentRunStore } from "../../mgr/store.js";
+import type { SelfTestCase, SelfTestContext } from "../harness.js";
+
+const policy: RunnerStartupRecoveryOptions = {
+  maxAttempts: 1,
+  initialBackoffMs: 1,
+  maxBackoffMs: 4,
+  multiplier: 2,
+  totalDeadlineMs: 1_000,
+  jitterRatio: 0,
+};
+
+const selfTest: SelfTestCase = async (context) => {
+  const kubectlCommand = path.join(context.root, "src/selftest/fixtures/fake-kubectl-runner-startup.mjs");
+  const fixturePath = path.join(context.tmp, "runner-startup-fixture.json");
+  const previousFixture = process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH;
+  process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH = fixturePath;
+  await chmod(kubectlCommand, 0o755);
+  try {
+    await assertTransientRegistryFailureExhausts(context, fixturePath, kubectlCommand);
+    await assertStartupRecoveryIsVisible(context, fixturePath, kubectlCommand);
+    await assertDeterministicFailureDoesNotRetry(context, fixturePath, kubectlCommand);
+  } finally {
+    if (previousFixture === undefined) delete process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH;
+    else process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH = previousFixture;
+  }
+  return {
+    name: "runner-startup-retry",
+    tests: [
+      "registry-unreachable-emits-bounded-exponential-retry",
+      "startup-recovery-emits-durable-recovered-event",
+      "image-not-found-fails-without-retry",
+      "semantic-failure-events-carry-workbench-fields",
+    ],
+  };
+};
+
+async function assertTransientRegistryFailureExhausts(context: SelfTestContext, fixturePath: string, kubectlCommand: string): Promise<void> {
+  const fact = createRunnerFact("transient");
+  await writeFixture(fixturePath, "ImagePullBackOff", "failed to pull image: dial tcp 10.43.249.83:5000: connect: connection refused");
+  await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: policy });
+  let events = fact.store.listEvents(fact.runId, 0, 100);
+  const observed = phase(events, "runner-startup-failure-observed");
+  const scheduled = phase(events, "runner-startup-retry-scheduled");
+  assert.equal(observed.payload.failureDomain, "infrastructure");
+  assert.equal(observed.payload.component, "runner-image-pull");
+  assert.equal(observed.payload.code, "runner-image-registry-unreachable");
+  assert.equal(observed.payload.retryable, true);
+  assert.equal(scheduled.payload.attempt, 1);
+  assert.equal(scheduled.payload.maxAttempts, 1);
+  assert.equal(scheduled.payload.backoffMs, 1);
+  assert.equal(typeof scheduled.payload.nextRetryAt, "string");
+  await Bun.sleep(5);
+  await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: policy });
+  events = fact.store.listEvents(fact.runId, 0, 100);
+  assert.ok(phase(events, "runner-startup-retry-started"));
+  assert.ok(phase(events, "runner-startup-retry-exhausted"));
+  assert.equal(fact.store.getCommand(fact.commandId).state, "failed");
+  assert.equal(fact.store.getRun(fact.runId).status, "failed");
+}
+
+async function assertStartupRecoveryIsVisible(context: SelfTestContext, fixturePath: string, kubectlCommand: string): Promise<void> {
+  const fact = createRunnerFact("recovered");
+  await writeFixture(fixturePath, "ErrImagePull", "failed to pull image: i/o timeout");
+  await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: { ...policy, maxAttempts: 3 } });
+  await writeFixture(fixturePath, "", "", "running");
+  await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: { ...policy, maxAttempts: 3 } });
+  const recovered = phase(fact.store.listEvents(fact.runId, 0, 100), "runner-startup-retry-recovered");
+  assert.equal(recovered.payload.failureDomain, "infrastructure");
+  assert.equal(recovered.payload.runnerJobId, fact.runnerJobId);
+  assert.equal(typeof recovered.payload.traceId, "string");
+  assert.equal(fact.store.getCommand(fact.commandId).state, "pending");
+}
+
+async function assertDeterministicFailureDoesNotRetry(context: SelfTestContext, fixturePath: string, kubectlCommand: string): Promise<void> {
+  const fact = createRunnerFact("not-found");
+  await writeFixture(fixturePath, "ErrImagePull", "failed to resolve reference: manifest unknown: not found");
+  await reconcileRunnerJobsOnce({ store: fact.store, namespace: "agentrun-v02", kubectlCommand, startupRecovery: { ...policy, maxAttempts: 5 } });
+  const events = fact.store.listEvents(fact.runId, 0, 100);
+  const exhausted = phase(events, "runner-startup-retry-exhausted");
+  assert.equal(exhausted.payload.code, "runner-image-not-found");
+  assert.equal(exhausted.payload.retryable, false);
+  assert.equal(events.some((event) => event.payload.phase === "runner-startup-retry-scheduled"), false);
+  assert.equal(fact.store.getRun(fact.runId).status, "failed");
+}
+
+function createRunnerFact(suffix: string): { store: MemoryAgentRunStore; runId: string; commandId: string; runnerJobId: string } {
+  const store = new MemoryAgentRunStore();
+  const run = store.createRun({
+    tenantId: "selftest",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: `/tmp/agentrun-startup-${suffix}` },
+    providerId: "NC01",
+    backendProfile: "codex",
+    executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } },
+    traceSink: null,
+  });
+  const command = store.createCommand(run.id, { type: "turn", payload: { prompt: suffix, traceId: `trc_startup_${suffix}` }, idempotencyKey: `startup-${suffix}` });
+  const runnerJob = store.saveRunnerJob({
+    runId: run.id,
+    commandId: command.id,
+    idempotencyKey: `startup-${suffix}`,
+    payloadHash: `hash-${suffix}`,
+    attemptId: `attempt-${suffix}`,
+    runnerId: `runner-${suffix}`,
+    namespace: "agentrun-v02",
+    jobName: `agentrun-v02-${suffix}`,
+    managerUrl: "http://agentrun-mgr",
+    image: "10.43.249.83:5000/agentrun/runner@sha256:selftest",
+    sourceCommit: "self-test",
+    serviceAccountName: "agentrun-v02-runner",
+    result: { valuesPrinted: false },
+  });
+  return { store, runId: run.id, commandId: command.id, runnerJobId: runnerJob.id };
+}
+
+async function writeFixture(fixturePath: string, reason: string, message: string, mode = "pending"): Promise<void> {
+  await writeFile(fixturePath, JSON.stringify({
+    namespace: "agentrun-v02",
+    createdAt: new Date(Date.now() - 1_000).toISOString(),
+    mode,
+    reason,
+    message,
+  }));
+}
+
+function phase(events: Array<{ payload: JsonRecord }>, name: string): { payload: JsonRecord } {
+  const event = events.find((item) => item.payload.phase === name);
+  assert.ok(event, `missing ${name}`);
+  return event;
+}
+
+export default selfTest;

src/selftest/cases/50-hwlab-manual-dispatch.ts

@@ -371,7 +371,9 @@ process.exit(1);
     ], "durable steers must be replayed exactly once per resumed backend attempt and keep command seq order");
     const retryEventsResponse = await client.get(`/api/v1/runs/${retryRun.runId}/events?afterSeq=0&limit=200`) as { items?: Array<{ type?: string; payload?: JsonRecord }> };
     const retryEvents = retryEventsResponse.items ?? [];
-    assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "runner:backend-retry").length, 1);
+    assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "runner-backend-retry-scheduled").length, 1);
+    assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "runner-backend-retry-started").length, 1);
+    assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "runner-backend-retry-recovered").length, 1);
     assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "thread/resume:completed").length, 2);
     const replayedSteers = retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:replayed");
     assert.deepEqual(replayedSteers.map((event) => event.payload?.commandId), [acknowledgedRetrySteer.id, firstRetrySteer.id, secondRetrySteer.id]);

src/selftest/fixtures/fake-kubectl-runner-startup.mjs

@@ -0,0 +1,59 @@
+#!/usr/bin/env node
+import { readFileSync } from "node:fs";
+
+const fixturePath = process.env.AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH;
+if (!fixturePath) {
+  process.stderr.write("AGENTRUN_SELFTEST_STARTUP_FIXTURE_PATH is required\n");
+  process.exit(2);
+}
+
+const fixture = JSON.parse(readFileSync(fixturePath, "utf8"));
+const args = process.argv.slice(2);
+const resource = args[1];
+if (args[0] !== "get") {
+  process.stderr.write(`unsupported command: ${args.join(" ")}\n`);
+  process.exit(2);
+}
+
+if (resource === "job") {
+  process.stdout.write(JSON.stringify({
+    kind: "Job",
+    metadata: {
+      name: args[2],
+      namespace: fixture.namespace,
+      uid: "job-startup-selftest",
+      creationTimestamp: fixture.createdAt,
+    },
+    status: { active: 1, startTime: fixture.createdAt },
+  }));
+  process.exit(0);
+}
+
+if (resource === "pods") {
+  const running = fixture.mode === "running";
+  const waiting = running ? undefined : {
+    reason: fixture.reason,
+    message: fixture.message,
+  };
+  process.stdout.write(JSON.stringify({
+    items: [{
+      kind: "Pod",
+      metadata: {
+        name: "runner-startup-selftest-pod",
+        namespace: fixture.namespace,
+        creationTimestamp: fixture.createdAt,
+      },
+      status: {
+        phase: running ? "Running" : "Pending",
+        containerStatuses: [{
+          name: "runner",
+          state: running ? { running: { startedAt: fixture.createdAt } } : { waiting },
+        }],
+      },
+    }],
+  }));
+  process.exit(0);
+}
+
+process.stderr.write(`unsupported resource: ${resource}\n`);
+process.exit(2);