From 06429f23959db95c7790f0ac3c4a04fd8826cdfd Mon Sep 17 00:00:00 2001 From: root Date: Sun, 12 Jul 2026 11:06:46 +0200 Subject: [PATCH] =?UTF-8?q?fix(mgr):=20=E5=9B=BA=E5=8C=96=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E5=A4=B1=E8=B4=A5=E4=BA=8B=E5=AE=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/kubernetes-runner-job.ts | 32 +++++------ src/mgr/postgres-store.ts | 17 +++--- src/mgr/runner-dispatcher.ts | 15 ++---- src/mgr/runner-retention.ts | 16 +++--- src/mgr/store.ts | 54 ++++++++++++++++--- .../cases/21-runner-durable-dispatch.ts | 47 +++++++++++++++- .../cases/66-session-turn-admission.ts | 10 +++- 7 files changed, 140 insertions(+), 51 deletions(-) diff --git a/src/mgr/kubernetes-runner-job.ts b/src/mgr/kubernetes-runner-job.ts index 3826b8e..a100fc4 100644 --- a/src/mgr/kubernetes-runner-job.ts +++ b/src/mgr/kubernetes-runner-job.ts @@ -255,17 +255,17 @@ async function createKubernetesRunnerJobUnderFence(options: { store: AgentRunSto let adopted = false; try { if (transientEnvSecretName) { - const secretCreate = await kubectlCreateOrAdopt(transientEnvSecretManifest({ namespace: render.namespace, name: transientEnvSecretName, runId: run.id, commandId, attemptId: render.attemptId, runnerId: render.runnerId, jobName: render.jobName, lane: lane ?? "v0.1", items: transientEnv }), kubectlCommand, "runner transient env secret", options.signal); + const secretCreate = await kubectlCreateOrAdopt(transientEnvSecretManifest({ namespace: render.namespace, name: transientEnvSecretName, runId: run.id, commandId, attemptId: render.attemptId, runnerId: render.runnerId, jobName: render.jobName, lane: lane ?? "v0.1", items: transientEnv }), kubectlCommand, "runner transient env secret", options.signal, { create: "runner-transient-env-secret-create-failed", adopt: "runner-transient-env-secret-adopt-failed" }); transientEnvSecretCreated = !secretCreate.adopted; } - const jobCreate = await kubectlCreateOrAdopt(render.manifest, kubectlCommand, "runner job", options.signal); + const jobCreate = await kubectlCreateOrAdopt(render.manifest, kubectlCommand, "runner job", options.signal, { create: "runner-kubernetes-job-create-failed", adopt: "runner-kubernetes-job-adopt-failed" }); created = jobCreate.object; adopted = jobCreate.adopted; } catch (error) { if (transientEnvSecretName && transientEnvSecretCreated) await kubectlDeleteSecret(transientEnvSecretName, render.namespace, kubectlCommand, options.signal); throw error; } - if (!created) throw new AgentRunError("infra-failed", "kubectl did not return created runner job metadata", { httpStatus: 502 }); + if (!created) throw new AgentRunError("infra-failed", "kubectl did not return created runner job metadata", { httpStatus: 502, details: { reason: "runner-kubernetes-job-create-failed", changeReason: "created-object-missing", valuesPrinted: false } }); throwIfAborted(options.signal, "runner dispatch attempt"); if (transientEnvSecretName) { const owner = await kubectlPatchSecretOwnerReference(transientEnvSecretName, render.namespace, { name: render.jobName, uid: objectPath(created, ["metadata", "uid"]) }, kubectlCommand, options.signal); @@ -518,7 +518,7 @@ function summarizeTransientEnvSecret(name: string, namespace: string, items: Run }; } -async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string, label = "runner job", abortSignal?: AbortSignal): Promise<{ object: JsonRecord; adopted: boolean }> { +async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string, label: string, abortSignal: AbortSignal | undefined, failureReasons: { create: string; adopt: string }): Promise<{ object: JsonRecord; adopted: boolean }> { throwIfAborted(abortSignal, `kubectl create ${label}`); const child = spawn(kubectlCommand, ["create", "-f", "-", "-o", "json"], { stdio: ["pipe", "pipe", "pipe"] }); let stdout = ""; @@ -537,26 +537,26 @@ async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string const name = optionalString(metadata.name); const namespace = optionalString(metadata.namespace); const kind = optionalString(manifest.kind)?.toLowerCase(); - if (!name || !namespace || !kind) throw new AgentRunError("infra-failed", `cannot adopt ${label} without deterministic identity`, { httpStatus: 502 }); + if (!name || !namespace || !kind) throw new AgentRunError("infra-failed", `cannot adopt ${label} without deterministic identity`, { httpStatus: 502, details: { reason: failureReasons.adopt, changeReason: "manifest-identity-missing", valuesPrinted: false } }); const existing = await kubectlRun(kubectlCommand, ["get", kind, name, "-n", namespace, "-o", "json"], abortSignal); - if (existing.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${label} failed during adopt`, { httpStatus: 502, details: { stderr: redactText(existing.stderr.slice(-2000)), valuesPrinted: false } }); - const object = parseKubectlObject(existing.stdout, `${label} adopt`); - assertAdoptedIdentity(manifest, object, label); + if (existing.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${label} failed during adopt`, { httpStatus: 502, details: { reason: failureReasons.adopt, changeReason: "existing-object-observation-failed", stderr: redactText(existing.stderr.slice(-2000)), valuesPrinted: false } }); + const object = parseKubectlObject(existing.stdout, `${label} adopt`, failureReasons.adopt); + assertAdoptedIdentity(manifest, object, label, failureReasons.adopt); return { object, adopted: true }; } - throw new AgentRunError("infra-failed", `kubectl create ${label} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(stderr.slice(-4000)), stdout: redactText(stdout.slice(-2000)), signal: result.signal }) }); + throw new AgentRunError("infra-failed", `kubectl create ${label} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ reason: failureReasons.create, exitCode: result.code, stderr: redactText(stderr.slice(-4000)), stdout: redactText(stdout.slice(-2000)), signal: result.signal, valuesPrinted: false }) }); } - return { object: parseKubectlObject(stdout, label), adopted: false }; + return { object: parseKubectlObject(stdout, label, failureReasons.create), adopted: false }; } -function parseKubectlObject(stdout: string, label: string): JsonRecord { +function parseKubectlObject(stdout: string, label: string, reason: string): JsonRecord { try { const parsed = JSON.parse(stdout) as unknown; if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) return kubernetesObjectSummary(parsed as JsonRecord); } catch (error) { - throw new AgentRunError("infra-failed", `kubectl returned invalid JSON for ${label}: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { stdoutHash: stableHash(stdout), valuesPrinted: false } }); + throw new AgentRunError("infra-failed", `kubectl returned invalid JSON for ${label}: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { reason, changeReason: "invalid-json", stdoutHash: stableHash(stdout), valuesPrinted: false } }); } - throw new AgentRunError("infra-failed", `kubectl returned non-object JSON for ${label}`, { httpStatus: 502 }); + throw new AgentRunError("infra-failed", `kubectl returned non-object JSON for ${label}`, { httpStatus: 502, details: { reason, changeReason: "non-object-json", stdoutHash: stableHash(stdout), valuesPrinted: false } }); } function kubernetesObjectSummary(object: JsonRecord): JsonRecord { @@ -575,17 +575,17 @@ function kubernetesObjectSummary(object: JsonRecord): JsonRecord { }); } -function assertAdoptedIdentity(expected: JsonRecord, actual: JsonRecord, label: string): void { +function assertAdoptedIdentity(expected: JsonRecord, actual: JsonRecord, label: string, reason: string): void { const expectedMetadata = objectRecord(expected.metadata); const actualMetadata = objectRecord(actual.metadata); for (const key of ["name", "namespace"] as const) { - if (expectedMetadata[key] !== actualMetadata[key]) throw new AgentRunError("infra-failed", `${label} adopt identity mismatch for ${key}`, { httpStatus: 409 }); + if (expectedMetadata[key] !== actualMetadata[key]) throw new AgentRunError("infra-failed", `${label} adopt identity mismatch for ${key}`, { httpStatus: 409, details: { reason, changeReason: "object-identity-mismatch", field: key, valuesPrinted: false } }); } const expectedAnnotations = objectRecord(expectedMetadata.annotations); const actualAnnotations = objectRecord(actualMetadata.annotations); for (const key of ["agentrun.pikastech.local/run-id", "agentrun.pikastech.local/command-id", "agentrun.pikastech.local/attempt-id", "agentrun.pikastech.local/runner-id", "agentrun.pikastech.local/runner-job-id"]) { if (expectedAnnotations[key] !== undefined && expectedAnnotations[key] !== actualAnnotations[key]) { - throw new AgentRunError("infra-failed", `${label} adopt annotation mismatch`, { httpStatus: 409, details: { annotation: key, valuesPrinted: false } }); + throw new AgentRunError("infra-failed", `${label} adopt annotation mismatch`, { httpStatus: 409, details: { reason, changeReason: "annotation-mismatch", annotation: key, valuesPrinted: false } }); } } } diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 6d58678..d0d56d5 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -6,7 +6,7 @@ import { redactJson } from "../common/redaction.js"; import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptListResult, QueueAttemptRecord, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueRetryActivation, QueueRetryReservation, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; import { newId, nowIso, stableHash } from "../common/validation.js"; import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, SessionTurnAdmission, SessionTurnAdmissionInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; -import { activeSessionAdmissionConflict, assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, assertSessionCommandReplay, assertSessionTurnAdmissionContract, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, invalidSessionAdmissionProjection, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, runCreatedEventPayload, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; +import { activeSessionAdmissionConflict, assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, assertSessionCommandReplay, assertSessionTurnAdmissionContract, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, invalidSessionAdmissionProjection, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, runCreatedEventPayload, runnerDispatchFailureKind, runnerDispatchFailureReason, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js"; import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js"; import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js"; @@ -744,7 +744,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( if (locked.fenced) return { intent, command, run, dispatchFenced: true, valuesPrinted: false }; const failure = redactJson(error); const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed"; - const reason = typeof failure.reason === "string" ? failure.reason : "runner-dispatch-attempt-failed"; + const failureKind = runnerDispatchFailureKind(failure.failureKind); + const reason = runnerDispatchFailureReason(failure.reason); const at = nowIso(); const updatedIntentResult = await client.query( `UPDATE agentrun_runner_dispatch_intents @@ -761,16 +762,16 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( let updatedRun = run; if (!isTerminalRunStatus(run.status)) { const updated = await client.query( - `UPDATE agentrun_runs SET status = 'failed', terminal_status = 'failed', failure_kind = 'infra-failed', failure_message = $2, updated_at = $3 + `UPDATE agentrun_runs SET status = 'failed', terminal_status = 'failed', failure_kind = $2, failure_message = $3, updated_at = $4 WHERE id = $1 RETURNING *`, - [run.id, message, at], + [run.id, failureKind, message, at], ); updatedRun = runFromRow(updated.rows[0]); await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, "run terminalized by runner dispatch failure", at); - await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); - await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message, threadId: null, turnId: null }); - await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message }); - await this.touchSessionForRun(client, updatedRun, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at }); + await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind, message, valuesPrinted: false }); + await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", reason, commandId: command.id, state: "failed", terminalStatus: "failed", failureKind, message, threadId: null, turnId: null }); + await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", reason, failureKind, message }); + await this.touchSessionForRun(client, updatedRun, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind, lastActivityAt: at }, { bumpVersion: true, at }); } return { intent: runnerDispatchIntentFromRow(updatedIntentResult.rows[0]), command: updatedCommand, run: updatedRun, valuesPrinted: false }; }); diff --git a/src/mgr/runner-dispatcher.ts b/src/mgr/runner-dispatcher.ts index 675a94a..158465e 100644 --- a/src/mgr/runner-dispatcher.ts +++ b/src/mgr/runner-dispatcher.ts @@ -2,7 +2,7 @@ import type { JsonRecord, RunnerDispatchCompletion, RunnerDispatchIntentRecord } import { AgentRunError } from "../common/errors.js"; import { redactJson, redactText } from "../common/redaction.js"; import { nowIso } from "../common/validation.js"; -import type { AgentRunStore } from "./store.js"; +import { runnerDispatchFailureKind, runnerDispatchFailureReason, type AgentRunStore } from "./store.js"; import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js"; import { childProcessKillGraceMs } from "./abortable-child.js"; @@ -57,7 +57,7 @@ export async function dispatchRunnerIntentsOnce(input: { await input.store.terminalizeRunnerDispatchIntent(intent, failure); failedCount++; } else { - await input.store.retryRunnerDispatchIntent(intent, failure, nextAttemptAt, { phase: "runner-dispatch-retry", reason: failure.reason, commandId: intent.commandId, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, nextAttemptAt, failureKind: "infra-failed", message: failure.message, valuesPrinted: false }); + await input.store.retryRunnerDispatchIntent(intent, failure, nextAttemptAt, { phase: "runner-dispatch-retry", reason: runnerDispatchFailureReason(failure.reason), commandId: intent.commandId, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, nextAttemptAt, failureKind: runnerDispatchFailureKind(failure.failureKind), message: failure.message, valuesPrinted: false }); retryCount++; } } catch (settleError) { @@ -102,8 +102,8 @@ function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): Js const message = redactText(error instanceof Error ? error.message : String(error)).slice(0, 500); const details = error instanceof AgentRunError ? redactJson(error.details ?? {}) : {}; return { - failureKind: error instanceof AgentRunError ? error.failureKind : "infra-failed", - reason: typeof details.reason === "string" ? details.reason : dispatchFailureReason(message), + failureKind: runnerDispatchFailureKind(error instanceof AgentRunError ? error.failureKind : null), + reason: runnerDispatchFailureReason(error instanceof AgentRunError ? details.reason : null), message, kubernetesObservation: details, dispatchIntentId: intent.id, @@ -115,13 +115,6 @@ function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): Js }; } -function dispatchFailureReason(message: string): string { - if (/kubectl get|retention.*observe|kubernetes.*observation/iu.test(message)) return "runner-kubernetes-observation-failed"; - if (/kubectl create|create runner job|runner job.*create/iu.test(message)) return "runner-kubernetes-job-create-failed"; - if (/aborted|attempt timeout|exceeded .* timeout/iu.test(message)) return "runner-dispatch-attempt-aborted"; - return "runner-dispatch-attempt-failed"; -} - function itemSummary(intent: RunnerDispatchIntentRecord, state: string, error: JsonRecord | null, completion?: RunnerDispatchCompletion): JsonRecord { return { dispatchIntentId: intent.id, runId: intent.runId, commandId: intent.commandId, runnerJobId: intent.runnerJobId, actualRunnerJobId: completion?.actualRunnerJobId ?? intent.actualRunnerJobId, activeRunnerId: completion?.activeRunnerId ?? intent.activeRunnerId, dispatchOutcome: completion?.dispatchOutcome ?? intent.dispatchOutcome, attemptCount: intent.attemptCount, state, failureKind: error?.failureKind ?? null, valuesPrinted: false }; } diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index e2b6a53..5b1cdcc 100644 --- a/src/mgr/runner-retention.ts +++ b/src/mgr/runner-retention.ts @@ -680,8 +680,8 @@ async function kubectlGetList(options: RunnerRetentionOptions, resource: string, if (selector) args.push("-l", selector); args.push("-o", "json"); const result = await kubectlRun(options.kubectlCommand ?? "kubectl", args, options.signal); - if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${resource} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) }); - const parsed = parseJsonObject(result.stdout, `kubectl get ${resource}`); + if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${resource} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ reason: "runner-kubernetes-observation-failed", namespace: options.namespace, resource, exitCode: result.code, signal: result.signal, stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) }); + const parsed = parseJsonObject(result.stdout, `kubectl get ${resource}`, "runner-kubernetes-observation-failed"); const items = parsed.items; return Array.isArray(items) ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) : []; } @@ -689,14 +689,14 @@ async function kubectlGetList(options: RunnerRetentionOptions, resource: string, async function kubectlGetObject(options: RunnerRetentionOptions, resource: "job" | "pod", name: string): Promise { const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["get", resource, name, "-n", options.namespace, "-o", "json"], options.signal); if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${resource}/${name} failed with code ${result.code}`, { httpStatus: 409, details: redactJson({ reason: "runner-retention-k8s-cas-rejected", changeReason: "resource-missing-before-cas", namespace: options.namespace, resourceKind: resource, name, stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) }); - return parseJsonObject(result.stdout, `kubectl get ${resource}/${name}`) as unknown as K8sObject; + return parseJsonObject(result.stdout, `kubectl get ${resource}/${name}`, "runner-retention-k8s-cas-rejected") as unknown as K8sObject; } async function kubectlGetPodEvents(options: RunnerRetentionOptions): Promise { const args = ["get", "events", "-n", options.namespace, "--field-selector", "involvedObject.kind=Pod,type=Warning", "-o", "json"]; const result = await kubectlRun(options.kubectlCommand ?? "kubectl", args, options.signal); - if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get events failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) }); - const parsed = parseJsonObject(result.stdout, "kubectl get events"); + if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get events failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ reason: "runner-kubernetes-observation-failed", namespace: options.namespace, resource: "events", exitCode: result.code, signal: result.signal, stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) }); + const parsed = parseJsonObject(result.stdout, "kubectl get events", "runner-kubernetes-observation-failed"); const items = parsed.items; return Array.isArray(items) ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) : []; } @@ -1155,14 +1155,14 @@ function stringPath(value: unknown, path: string[]): string | null { return typeof current === "string" ? current : null; } -function parseJsonObject(text: string, label: string): JsonRecord { +function parseJsonObject(text: string, label: string, reason: string): JsonRecord { try { const parsed = JSON.parse(text) as JsonValue; if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) return parsed; } catch (error) { - throw new AgentRunError("infra-failed", `${label} returned invalid JSON: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { stdoutHash: stableHash(text), valuesPrinted: false } }); + throw new AgentRunError("infra-failed", `${label} returned invalid JSON: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { reason, stdoutHash: stableHash(text), valuesPrinted: false } }); } - throw new AgentRunError("infra-failed", `${label} returned non-object JSON`, { httpStatus: 502, details: { stdoutHash: stableHash(text), valuesPrinted: false } }); + throw new AgentRunError("infra-failed", `${label} returned non-object JSON`, { httpStatus: 502, details: { reason, stdoutHash: stableHash(text), valuesPrinted: false } }); } function countDeletedLines(stdout: string): number { diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 84dee46..7442ed9 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -462,20 +462,21 @@ export class MemoryAgentRunStore implements AgentRunStore { const at = nowIso(); const failure = redactJson(error); const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed"; - const reason = typeof failure.reason === "string" ? failure.reason : "runner-dispatch-attempt-failed"; + const failureKind = runnerDispatchFailureKind(failure.failureKind); + const reason = runnerDispatchFailureReason(failure.reason); const updatedIntent: RunnerDispatchIntentRecord = { ...intent, state: "failed", leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: at, lastError: failure, updatedAt: at }; this.runnerDispatchIntents.set(intent.id, updatedIntent); const command = this.getCommand(intent.commandId); const updatedCommand = isTerminalCommandState(command.state) ? command : { ...command, state: "failed" as const, updatedAt: at }; this.commands.set(command.id, updatedCommand); const run = this.getRun(intent.runId); - const updatedRun = isTerminalRunStatus(run.status) ? run : this.updateRun(run.id, { status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: message }); + const updatedRun = isTerminalRunStatus(run.status) ? run : this.updateRun(run.id, { status: "failed", terminalStatus: "failed", failureKind, failureMessage: message }); if (!isTerminalRunStatus(run.status)) { this.cancelRunnerDispatchIntentsForRun(run.id, "run terminalized by runner dispatch failure", at); - this.appendEvent(run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); - this.appendEvent(run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message, threadId: null, turnId: null }); - this.appendEvent(run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message }); - this.touchSessionForRun(this.getRun(run.id), { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at }); + this.appendEvent(run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind, message, valuesPrinted: false }); + this.appendEvent(run.id, "backend_status", { phase: "command-terminal", reason, commandId: command.id, state: "failed", terminalStatus: "failed", failureKind, message, threadId: null, turnId: null }); + this.appendEvent(run.id, "terminal_status", { terminalStatus: "failed", reason, failureKind, message }); + this.touchSessionForRun(this.getRun(run.id), { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind, lastActivityAt: at }, { bumpVersion: true, at }); } return { intent: updatedIntent, command: updatedCommand, run: updatedRun, valuesPrinted: false }; } @@ -1337,6 +1338,47 @@ export class MemoryAgentRunStore implements AgentRunStore { } } +export function runnerDispatchFailureKind(value: unknown): FailureKind { + switch (value) { + case "auth-missing": + case "auth-failed": + case "schema-invalid": + case "tenant-policy-denied": + case "secret-unavailable": + case "prompt-unavailable": + case "prompt-too-large": + case "required-skill-unavailable": + case "skill-unavailable": + case "runner-lease-conflict": + case "backend-failed": + case "backend-protocol-error": + case "backend-spawn-failed": + case "backend-json-parse-error": + case "backend-response-invalid": + case "thread-resume-failed": + case "backend-timeout": + case "provider-auth-failed": + case "provider-rate-limited": + case "provider-stream-disconnected": + case "provider-http-error": + case "provider-invalid-tool-call": + case "provider-compact-unsupported": + case "provider-unavailable": + case "infra-failed": + case "session-store-evicted": + case "cancelled": + return value; + default: + return "infra-failed"; + } +} + +export function runnerDispatchFailureReason(value: unknown): string { + if (typeof value !== "string") return "runner-dispatch-attempt-failed"; + const reason = value.trim(); + return reason.length <= 120 && /^[a-z0-9]+(?:-[a-z0-9]+)*$/u.test(reason) ? reason : "runner-dispatch-attempt-failed"; +} + export function assertSessionBoundary(existing: SessionRecord, input: CreateRunInput): void { if (existing.tenantId !== input.tenantId || existing.projectId !== input.projectId) { throw new AgentRunError("tenant-policy-denied", "sessionRef cannot be reused across tenant or project boundary", { httpStatus: 403, details: { sessionId: existing.sessionId, valuesPrinted: false } }); diff --git a/src/selftest/cases/21-runner-durable-dispatch.ts b/src/selftest/cases/21-runner-durable-dispatch.ts index b36c6b2..3d4ef33 100644 --- a/src/selftest/cases/21-runner-durable-dispatch.ts +++ b/src/selftest/cases/21-runner-durable-dispatch.ts @@ -2,6 +2,7 @@ import assert from "node:assert/strict"; import { access, chmod, rm } from "node:fs/promises"; import path from "node:path"; import type { CreateRunInput, RunEvent, RunRecord } from "../../common/types.js"; +import { AgentRunError } from "../../common/errors.js"; import { validateCreateCommand } from "../../common/validation.js"; import { dispatchRunnerIntentsOnce, type RunnerDispatcherOptions } from "../../mgr/runner-dispatcher.js"; import { MemoryAgentRunStore } from "../../mgr/store.js"; @@ -49,6 +50,7 @@ const selfTest: SelfTestCase = async (context) => { await assertEnabledDispatcherRejectsLegacyDefaults(); assertTerminalTransitionsFenceDispatchIntents(); assertSettleFencesNewlyTerminalParent(); + await assertTypedDispatchFailureFacts(fakeKubectl); process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "fail"; const failedRun = store.createRun({ ...runInput(), sessionRef: null }); @@ -67,7 +69,7 @@ const selfTest: SelfTestCase = async (context) => { restoreEnv("AGENTRUN_SELFTEST_KUBECTL_STATE", previousState); restoreEnv("AGENTRUN_SELFTEST_KUBECTL_MODE", previousMode); } - return { name: "runner-durable-dispatch", tests: ["http-admission-contract", "orphan-job-adopt", "active-runner-reuse", "attempt-timeout-recovery", "just-in-time-claim", "atomic-settle-event", "enabled-config-fail-fast", "terminal-intent-fencing", "settle-secondary-fence", "permanent-failure-terminal"] }; + return { name: "runner-durable-dispatch", tests: ["http-admission-contract", "orphan-job-adopt", "active-runner-reuse", "attempt-timeout-recovery", "just-in-time-claim", "atomic-settle-event", "enabled-config-fail-fast", "terminal-intent-fencing", "settle-secondary-fence", "typed-dispatch-failure-facts", "permanent-failure-terminal"] }; }; async function assertHttpAdmissionContract(kubectlCommand: string): Promise { @@ -264,6 +266,49 @@ class TerminalProjectionStore extends MemoryAgentRunStore { } } +async function assertTypedDispatchFailureFacts(kubectlCommand: string): Promise { + const typedRetryStore = new RunnerCreateFailureStore(new AgentRunError("backend-timeout", "typed runner dispatch timeout", { httpStatus: 504, details: { reason: "runner-dispatch-source-timeout", valuesPrinted: false } })); + const typedRetryRun = typedRetryStore.createRun(runInput()); + const typedRetryCommand = typedRetryStore.createCommand(typedRetryRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); + const typedRetry = await dispatchRunnerIntentsOnce({ store: typedRetryStore, defaults: defaults(kubectlCommand), options: dispatcherOptions(3) }); + assert.equal(typedRetry.retryCount, 1); + assert.equal(typedRetryStore.getRunnerDispatchIntent(typedRetryCommand.id)?.lastError?.failureKind, "backend-timeout"); + assert.equal(typedRetryStore.getRunnerDispatchIntent(typedRetryCommand.id)?.lastError?.reason, "runner-dispatch-source-timeout"); + const retryEvent = typedRetryStore.listEvents(typedRetryRun.id, 0, 100).find((event) => event.payload.phase === "runner-dispatch-retry"); + assert.equal(retryEvent?.payload.failureKind, "backend-timeout"); + assert.equal(retryEvent?.payload.reason, "runner-dispatch-source-timeout"); + + const typedTerminalStore = new RunnerCreateFailureStore(new AgentRunError("backend-timeout", "typed runner dispatch timeout", { httpStatus: 504, details: { reason: "runner-dispatch-source-timeout", valuesPrinted: false } })); + const typedTerminalRun = typedTerminalStore.createRun(runInput()); + const typedTerminalCommand = typedTerminalStore.createCommand(typedTerminalRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); + const typedTerminal = await dispatchRunnerIntentsOnce({ store: typedTerminalStore, defaults: defaults(kubectlCommand), options: dispatcherOptions(1) }); + assert.equal(typedTerminal.failedCount, 1); + assert.equal(typedTerminalStore.getRunnerDispatchIntent(typedTerminalCommand.id)?.lastError?.failureKind, "backend-timeout"); + assert.equal(typedTerminalStore.getRun(typedTerminalRun.id).failureKind, "backend-timeout"); + const typedTerminalEvents = typedTerminalStore.listEvents(typedTerminalRun.id, 0, 100).filter((item) => item.payload.phase === "runner-dispatch-failed" || item.payload.phase === "command-terminal" || item.type === "terminal_status"); + assert.equal(typedTerminalEvents.length, 3); + for (const event of typedTerminalEvents) { + assert.equal(event.payload.failureKind, "backend-timeout"); + assert.equal(event.payload.reason, "runner-dispatch-source-timeout"); + } + + const untypedStore = new RunnerCreateFailureStore(new Error("kubectl get jobs then kubectl create runner job failed")); + const untypedRun = untypedStore.createRun(runInput()); + const untypedCommand = untypedStore.createCommand(untypedRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); + const untyped = await dispatchRunnerIntentsOnce({ store: untypedStore, defaults: defaults(kubectlCommand), options: dispatcherOptions(1) }); + assert.equal(untyped.failedCount, 1); + assert.equal(untypedStore.getRunnerDispatchIntent(untypedCommand.id)?.lastError?.failureKind, "infra-failed"); + assert.equal(untypedStore.getRunnerDispatchIntent(untypedCommand.id)?.lastError?.reason, "runner-dispatch-attempt-failed"); +} + +class RunnerCreateFailureStore extends MemoryAgentRunStore { + constructor(private readonly failure: unknown) { super(); } + + override async withRunnerCreateFence(_fenceKey: string, _operation: () => Promise): Promise { + throw this.failure; + } +} + async function waitFor(predicate: () => boolean, timeoutMs: number, message: string | (() => string)): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { diff --git a/src/selftest/cases/66-session-turn-admission.ts b/src/selftest/cases/66-session-turn-admission.ts index 41a0882..5374cb5 100644 --- a/src/selftest/cases/66-session-turn-admission.ts +++ b/src/selftest/cases/66-session-turn-admission.ts @@ -282,15 +282,23 @@ async function assertSuccessfulSameSessionContinuity(kubectlCommand: string): Pr function assertTerminalizedAdmission(store: MemoryAgentRunStore, runId: string, commandId: string, sessionId: string, reason: string): void { const intent = store.getRunnerDispatchIntent(commandId); assert.equal(intent?.state, "failed"); + assert.equal(intent?.lastError?.failureKind, "infra-failed"); assert.equal(intent?.lastError?.reason, reason); assert.equal(intent?.lastError?.valuesPrinted, false); assert.equal(store.getCommand(commandId).state, "failed"); assert.equal(store.getRun(runId).status, "failed"); + assert.equal(store.getRun(runId).failureKind, "infra-failed"); const session = store.getSession(sessionId); assert.equal(session?.activeRunId, null); assert.equal(session?.activeCommandId, null); assert.equal(session?.executionState, "terminal"); - assert.equal(store.listEvents(runId, 0, 100).find((event) => event.payload.phase === "runner-dispatch-failed")?.payload.reason, reason); + assert.equal(session?.failureKind, "infra-failed"); + const terminalEvents = store.listEvents(runId, 0, 100).filter((item) => item.payload.phase === "runner-dispatch-failed" || item.payload.phase === "command-terminal" || item.type === "terminal_status"); + assert.equal(terminalEvents.length, 3); + for (const event of terminalEvents) { + assert.equal(event.payload.failureKind, "infra-failed"); + assert.equal(event.payload.reason, reason); + } assert.equal(store.listEvents(runId, 0, 100).filter((event) => event.type === "terminal_status").length, 1); }