fix(mgr): 固化调度失败事实
This commit is contained in:
root
@@ -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 } });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 };
|
||||
});
|
||||
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
@@ -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<K8sObject> {
|
||||
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<K8sObject[]> {
|
||||
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 {
|
||||
|
||||
+48
-6
@@ -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 } });
|
||||
|
||||
@@ -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<void> {
|
||||
@@ -264,6 +266,49 @@ class TerminalProjectionStore extends MemoryAgentRunStore {
|
||||
}
|
||||
}
|
||||
|
||||
async function assertTypedDispatchFailureFacts(kubectlCommand: string): Promise<void> {
|
||||
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<T>(_fenceKey: string, _operation: () => Promise<T>): Promise<T> {
|
||||
throw this.failure;
|
||||
}
|
||||
}
|
||||
|
||||
async function waitFor(predicate: () => boolean, timeoutMs: number, message: string | (() => string)): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user