From e629a46e448c2c7a4759082506aa0118384b9921 Mon Sep 17 00:00:00 2001 From: AgentRun Codex Date: Mon, 20 Jul 2026 06:24:08 +0200 Subject: [PATCH] fix: expose bounded runner recovery events --- src/mgr/runner-job-status.ts | 2 + src/mgr/runner-reconciler.ts | 317 +++++++++++++++++- src/mgr/server.ts | 52 ++- src/runner/run-once.ts | 102 +++++- src/selftest/cases/24-runner-startup-retry.ts | 138 ++++++++ .../cases/50-hwlab-manual-dispatch.ts | 4 +- .../fixtures/fake-kubectl-runner-startup.mjs | 59 ++++ 7 files changed, 661 insertions(+), 13 deletions(-) create mode 100644 src/selftest/cases/24-runner-startup-retry.ts create mode 100755 src/selftest/fixtures/fake-kubectl-runner-startup.mjs diff --git a/src/mgr/runner-job-status.ts b/src/mgr/runner-job-status.ts index a518fee..5dddd58 100644 --- a/src/mgr/runner-job-status.ts +++ b/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, diff --git a/src/mgr/runner-reconciler.ts b/src/mgr/runner-reconciler.ts index 1de89ad..f52d85c 100644 --- a/src/mgr/runner-reconciler.ts +++ b/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 { + 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 { + 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)[key]; + } + return Array.isArray(current) ? current : []; +} + function jobCondition(jobObject: K8sObject | null): string | null { const conditions = jobObject?.status?.conditions; if (!Array.isArray(conditions)) return null; diff --git a/src/mgr/server.ts b/src/mgr/server.ts index 5ff70dc..c344d26 100644 --- a/src/mgr/server.ts +++ b/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 { 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; diff --git a/src/runner/run-once.ts b/src/runner/run-once.ts index 763b28d..9af4c74 100644 --- a/src/runner/run-once.ts +++ b/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); diff --git a/src/selftest/cases/24-runner-startup-retry.ts b/src/selftest/cases/24-runner-startup-retry.ts new file mode 100644 index 0000000..5fb4ec0 --- /dev/null +++ b/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 { + 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 { + 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 { + 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 { + 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; diff --git a/src/selftest/cases/50-hwlab-manual-dispatch.ts b/src/selftest/cases/50-hwlab-manual-dispatch.ts index 304aec1..bb50bc1 100644 --- a/src/selftest/cases/50-hwlab-manual-dispatch.ts +++ b/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]); diff --git a/src/selftest/fixtures/fake-kubectl-runner-startup.mjs b/src/selftest/fixtures/fake-kubectl-runner-startup.mjs new file mode 100755 index 0000000..1645765 --- /dev/null +++ b/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);