Lyon
665 lines
29 KiB
TypeScript
665 lines
29 KiB
TypeScript
import { spawn } from "node:child_process";
|
|
import { AgentRunError } from "../common/errors.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 { nowIso, stableHash } from "../common/validation.js";
|
|
import type { AgentRunStore } from "./store.js";
|
|
import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
|
|
|
|
export interface RunnerReconcilerOptions {
|
|
store: AgentRunStore;
|
|
namespace?: string;
|
|
kubectlCommand?: string;
|
|
limit?: number;
|
|
}
|
|
|
|
export interface RunnerReconcilerLoopOptions extends RunnerReconcilerOptions {
|
|
batchSize: number;
|
|
intervalMs: number;
|
|
onError?: (error: unknown) => void;
|
|
}
|
|
|
|
interface K8sObject {
|
|
kind?: string;
|
|
metadata?: {
|
|
name?: string;
|
|
namespace?: string;
|
|
uid?: string;
|
|
creationTimestamp?: string;
|
|
labels?: Record<string, string>;
|
|
annotations?: Record<string, string>;
|
|
ownerReferences?: Array<{ kind?: string; name?: string }>;
|
|
};
|
|
status?: JsonRecord;
|
|
}
|
|
|
|
interface K8sList {
|
|
items?: K8sObject[];
|
|
}
|
|
|
|
interface CommandTerminalReport {
|
|
terminalStatus: TerminalStatus;
|
|
failureKind: FailureKind | null;
|
|
failureMessage: string | null;
|
|
threadId?: string;
|
|
turnId?: string;
|
|
}
|
|
|
|
interface TerminalOutboxRecord {
|
|
outboxKey: string;
|
|
phase: string | null;
|
|
runId: string;
|
|
commandId: string;
|
|
runnerId: string;
|
|
runnerJobId: string | null;
|
|
attemptId: string;
|
|
report: CommandTerminalReport;
|
|
terminalRun: boolean;
|
|
events: BackendEvent[];
|
|
}
|
|
|
|
interface TerminalOutboxSearch {
|
|
outbox: TerminalOutboxRecord | null;
|
|
lineCount: number;
|
|
candidateCount: number;
|
|
parseFailedCount: number;
|
|
stdoutHash: string;
|
|
}
|
|
|
|
export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): Promise<JsonRecord> {
|
|
const limit = clampLimit(input.limit ?? 20);
|
|
const jobs = await input.store.listRunnerJobsForReconciliation(limit);
|
|
const items: JsonRecord[] = [];
|
|
let observeFailedCount = 0;
|
|
let runClosureCount = 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 });
|
|
if (stringValue(observation.category) === "runner-job-observe-failed") observeFailedCount++;
|
|
const runClosure = await closeOpenRunWhenCommandTerminal(input.store, job, observation);
|
|
if (runClosure.closed === true) runClosureCount++;
|
|
items.push({
|
|
runnerJobId: job.id,
|
|
runId: job.runId,
|
|
commandId: job.commandId,
|
|
namespace: stringValue(observation.namespace) ?? job.namespace,
|
|
jobName: job.jobName,
|
|
observedRunnerPhase: stringValue(observation.observedRunnerPhase) ?? "unknown",
|
|
terminalReportState: stringValue(observation.terminalReportState) ?? "unknown",
|
|
runReportState: stringValue(observation.runReportState) ?? "unknown",
|
|
terminalOutboxState: terminalOutboxState(observation),
|
|
runClosure,
|
|
valuesPrinted: false,
|
|
});
|
|
}
|
|
|
|
return {
|
|
action: "runner-job-reconcile",
|
|
reconciledAt: nowIso(),
|
|
limit,
|
|
scannedCount: jobs.length,
|
|
updatedCount: jobs.length,
|
|
observeFailedCount,
|
|
runClosureCount,
|
|
items,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
async function recoverTerminalOutboxIfNeeded(store: AgentRunStore, job: RunnerJobRecord, observation: JsonRecord, input: { namespace?: string; kubectlCommand?: string }): Promise<JsonRecord | null> {
|
|
const phase = stringValue(observation.observedRunnerPhase);
|
|
if (phase !== "k8s:succeeded" && phase !== "k8s:failed") return null;
|
|
try {
|
|
const command = await store.getCommand(job.commandId);
|
|
if (isTerminalCommandState(command.state)) {
|
|
return { state: "command-terminal", terminalReportState: "terminal-command-already-recorded", commandState: command.state, valuesPrinted: false };
|
|
}
|
|
const namespace = input.namespace ?? job.namespace;
|
|
const kubectlCommand = input.kubectlCommand ?? "kubectl";
|
|
const logs = await getRunnerJobLogs(kubectlCommand, namespace, job.jobName);
|
|
if (logs.code !== 0) {
|
|
return {
|
|
state: "logs-unavailable",
|
|
terminalReportState: "terminal-outbox-unavailable",
|
|
runReportState: "not-updated",
|
|
failureKind: "infra-failed",
|
|
failureMessage: redactText(logs.stderr || `kubectl logs failed with code ${logs.code}`).slice(0, 300),
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
const search = findLatestTerminalOutbox(logs.stdout, job);
|
|
if (!search.outbox) {
|
|
return {
|
|
state: "not-found",
|
|
terminalReportState: "terminal-outbox-missing",
|
|
runReportState: "not-updated",
|
|
lineCount: search.lineCount,
|
|
candidateCount: search.candidateCount,
|
|
parseFailedCount: search.parseFailedCount,
|
|
stdoutHash: search.stdoutHash,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
const beforeCommand = await getCommandOrNull(store, job.commandId);
|
|
const eventReplay = await replayTerminalOutboxEvents(store, search.outbox);
|
|
const nextCommand = await store.finishCommand(job.commandId, search.outbox.report);
|
|
const runRecovery = await recoverTerminalOutboxRun(store, search.outbox);
|
|
const otel = await emitTerminalOutboxOtelSpans(store, search.outbox, nextCommand, {
|
|
commandTransitioned: beforeCommand === null || (!isTerminalCommandState(beforeCommand.state) && isTerminalCommandState(nextCommand.state)),
|
|
runTransitioned: stringValue(runRecovery.state) === "terminal-outbox-replayed",
|
|
});
|
|
return {
|
|
state: "recovered",
|
|
terminalReportState: "terminal-outbox-replayed",
|
|
runReportState: stringValue(runRecovery.state) ?? "not-updated",
|
|
outboxKey: search.outbox.outboxKey,
|
|
phase: search.outbox.phase,
|
|
commandState: nextCommand.state,
|
|
terminalStatus: search.outbox.report.terminalStatus,
|
|
failureKind: search.outbox.report.failureKind,
|
|
failureMessageHash: search.outbox.report.failureMessage ? stableHash({ message: search.outbox.report.failureMessage }) : null,
|
|
terminalRun: search.outbox.terminalRun,
|
|
lineCount: search.lineCount,
|
|
candidateCount: search.candidateCount,
|
|
parseFailedCount: search.parseFailedCount,
|
|
eventReplay,
|
|
runRecovery,
|
|
otel,
|
|
valuesPrinted: false,
|
|
};
|
|
} catch (error) {
|
|
return {
|
|
state: "recovery-failed",
|
|
terminalReportState: "terminal-outbox-recovery-failed",
|
|
runReportState: "not-updated",
|
|
failureKind: "infra-failed",
|
|
failureMessage: error instanceof Error ? redactText(error.message).slice(0, 300) : "unknown terminal outbox recovery failure",
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
}
|
|
|
|
async function getCommandOrNull(store: AgentRunStore, commandId: string): Promise<CommandRecord | null> {
|
|
try {
|
|
return await store.getCommand(commandId);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
async function emitTerminalOutboxOtelSpans(store: AgentRunStore, outbox: TerminalOutboxRecord, command: CommandRecord, state: { commandTransitioned: boolean; runTransitioned: boolean }): Promise<JsonRecord> {
|
|
if (!state.commandTransitioned && !state.runTransitioned) return { emitted: false, reason: "already-terminal", valuesPrinted: false };
|
|
const run = await store.getRun(outbox.runId);
|
|
const common = {
|
|
source: "manager-reconciler",
|
|
phase: "command-terminal",
|
|
outboxKey: outbox.outboxKey,
|
|
outboxPhase: outbox.phase,
|
|
terminalStatus: outbox.report.terminalStatus,
|
|
failureKind: outbox.report.failureKind,
|
|
commandId: outbox.commandId,
|
|
attemptId: outbox.attemptId,
|
|
runnerId: outbox.runnerId,
|
|
runnerJobId: outbox.runnerJobId,
|
|
terminalRun: outbox.terminalRun === true,
|
|
commandTransitioned: state.commandTransitioned,
|
|
runTransitioned: state.runTransitioned,
|
|
};
|
|
const commandSpan = await emitAgentRunOtelSpan("runner_command_terminal", run, process.env, {
|
|
command,
|
|
kind: 2,
|
|
status: outbox.report.terminalStatus === "failed" || outbox.report.terminalStatus === "blocked" ? "error" : "ok",
|
|
error: outbox.report.terminalStatus === "failed" || outbox.report.terminalStatus === "blocked" ? outbox.report.failureMessage ?? outbox.report.failureKind ?? "terminal failure" : undefined,
|
|
attributes: common,
|
|
});
|
|
const terminalSpan = command.type === "turn"
|
|
? await emitAgentRunOtelSpan(`runner_terminal.${outbox.report.terminalStatus}`, run, process.env, {
|
|
command,
|
|
kind: 2,
|
|
status: outbox.report.terminalStatus === "failed" || outbox.report.terminalStatus === "blocked" ? "error" : "ok",
|
|
error: outbox.report.terminalStatus === "failed" || outbox.report.terminalStatus === "blocked" ? outbox.report.failureMessage ?? outbox.report.failureKind ?? "terminal failure" : undefined,
|
|
attributes: { ...common, eventType: "terminal_status" },
|
|
})
|
|
: { skipped: true, reason: "non-turn-command", valuesPrinted: false };
|
|
return { emitted: true, commandSpan, terminalSpan, valuesPrinted: false };
|
|
}
|
|
|
|
async function replayTerminalOutboxEvents(store: AgentRunStore, outbox: TerminalOutboxRecord): Promise<JsonRecord> {
|
|
const existing = await store.listEvents(outbox.runId, 0, 1_000);
|
|
const existingKeys = new Set(existing.map((event) => eventContentKey({ type: event.type, payload: event.payload })));
|
|
let appendedCount = 0;
|
|
let skippedCount = 0;
|
|
for (const event of outbox.events) {
|
|
const key = eventContentKey(event);
|
|
if (existingKeys.has(key)) {
|
|
skippedCount++;
|
|
continue;
|
|
}
|
|
await store.appendEvent(outbox.runId, event.type, event.payload);
|
|
existingKeys.add(key);
|
|
appendedCount++;
|
|
}
|
|
return { eventCount: outbox.events.length, appendedCount, skippedCount, valuesPrinted: false };
|
|
}
|
|
|
|
async function recoverTerminalOutboxRun(store: AgentRunStore, outbox: TerminalOutboxRecord): Promise<JsonRecord> {
|
|
if (outbox.terminalRun !== true) return { state: "not-requested", valuesPrinted: false };
|
|
const run = await store.getRun(outbox.runId);
|
|
if (isTerminalRunStatus(run.status)) return { state: "already-terminal", runStatus: run.status, valuesPrinted: false };
|
|
const next = await store.finishRun(outbox.runId, outbox.report);
|
|
return { state: "terminal-outbox-replayed", runStatus: next.status, terminalStatus: outbox.report.terminalStatus, valuesPrinted: false };
|
|
}
|
|
|
|
function mergeTerminalOutboxObservation(observation: JsonRecord, terminalOutbox: JsonRecord): JsonRecord {
|
|
return {
|
|
...observation,
|
|
terminalOutbox,
|
|
terminalReportState: stringValue(terminalOutbox.terminalReportState) ?? stringValue(observation.terminalReportState) ?? "unknown",
|
|
runReportState: stringValue(terminalOutbox.runReportState) ?? stringValue(observation.runReportState) ?? "unknown",
|
|
evidenceLevel: terminalOutbox.state === "recovered" ? "high" : observation.evidenceLevel ?? "medium",
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function terminalOutboxState(observation: JsonRecord): string {
|
|
return stringValue(recordValue(observation.terminalOutbox)?.state) ?? "not-checked";
|
|
}
|
|
|
|
export function startRunnerJobReconciler(input: RunnerReconcilerLoopOptions): () => void {
|
|
const intervalMs = Math.max(1_000, input.intervalMs);
|
|
let stopped = false;
|
|
let running = false;
|
|
const tick = async (): Promise<void> => {
|
|
if (stopped || running) return;
|
|
running = true;
|
|
try {
|
|
await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) });
|
|
} catch (error) {
|
|
input.onError?.(error);
|
|
} finally {
|
|
running = false;
|
|
}
|
|
};
|
|
const timer = setInterval(() => { void tick(); }, intervalMs);
|
|
timer.unref?.();
|
|
void tick();
|
|
return () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
};
|
|
}
|
|
|
|
async function observeRunnerJob(job: RunnerJobRecord, input: { namespace?: string; kubectlCommand?: string }): Promise<JsonRecord> {
|
|
const namespace = input.namespace ?? job.namespace;
|
|
const kubectlCommand = input.kubectlCommand ?? "kubectl";
|
|
const observedAt = nowIso();
|
|
try {
|
|
const [jobObject, podObjects] = await Promise.all([
|
|
getK8sObject(kubectlCommand, "job", namespace, job.jobName),
|
|
getK8sList(kubectlCommand, "pods", namespace, ["-l", `job-name=${job.jobName}`]),
|
|
]);
|
|
return observationFromObjects(job, namespace, observedAt, jobObject, podObjects);
|
|
} catch (error) {
|
|
return {
|
|
source: "manager-reconciler",
|
|
namespace,
|
|
jobName: job.jobName,
|
|
observedRunnerPhase: "k8s:observe-failed",
|
|
category: "runner-job-observe-failed",
|
|
terminalStatus: null,
|
|
failureKind: "infra-failed",
|
|
failureMessage: error instanceof Error ? redactText(error.message).slice(0, 300) : "unknown observation failure",
|
|
terminalReportState: "k8s-observation-failed",
|
|
runReportState: "not-updated",
|
|
lastK8sObservedAt: observedAt,
|
|
evidenceLevel: "medium",
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
}
|
|
|
|
function observationFromObjects(job: RunnerJobRecord, namespace: string, observedAt: string, jobObject: K8sObject | null, podObjects: K8sObject[]): JsonRecord {
|
|
const phase = observedRunnerPhase(jobObject, podObjects);
|
|
const terminalReportState = terminalReportStateForPhase(phase);
|
|
const failureKind: FailureKind | null = phase === "k8s:failed" || phase === "k8s:missing" ? "infra-failed" : null;
|
|
const k8s = k8sSummary(jobObject, podObjects);
|
|
return {
|
|
source: "manager-reconciler",
|
|
namespace,
|
|
jobName: job.jobName,
|
|
observedRunnerPhase: phase,
|
|
phase,
|
|
category: categoryForPhase(phase),
|
|
terminalStatus: null,
|
|
failureKind,
|
|
terminalReportState,
|
|
runReportState: "not-updated",
|
|
lastK8sObservedAt: observedAt,
|
|
lastObservedAt: observedAt,
|
|
lastObservedKind: `manager-reconciler:${phase}`,
|
|
evidenceLevel: phase === "k8s:unknown" ? "medium" : "high",
|
|
k8s,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
async function closeOpenRunWhenCommandTerminal(store: AgentRunStore, job: RunnerJobRecord, observation: JsonRecord): Promise<JsonRecord> {
|
|
let command: CommandRecord;
|
|
try {
|
|
command = await store.getCommand(job.commandId);
|
|
} catch {
|
|
return { closed: false, state: "command-missing", valuesPrinted: false };
|
|
}
|
|
if (!isTerminalCommandState(command.state)) return { closed: false, state: "command-open", commandState: command.state, valuesPrinted: false };
|
|
const observedRunnerPhase = stringValue(observation.observedRunnerPhase) ?? stringValue(observation.phase);
|
|
if (!runnerJobAllowsRunClosure(observedRunnerPhase)) return { closed: false, state: "runner-job-still-active", observedRunnerPhase: observedRunnerPhase ?? "unknown", commandState: command.state, valuesPrinted: false };
|
|
const run = await store.getRun(job.runId);
|
|
if (isTerminalRunStatus(run.status)) return { closed: false, state: "run-terminal", runStatus: run.status, commandState: command.state, valuesPrinted: false };
|
|
const terminalStatus = terminalStatusFromCommand(command);
|
|
const failureKind: FailureKind | null = terminalStatus === "cancelled" ? "cancelled" : terminalStatus === "failed" ? "infra-failed" : null;
|
|
const failureMessage = terminalStatus === "completed" ? null : "manager reconciler closed open run after terminal command state";
|
|
const next = await store.finishRun(run.id, { terminalStatus, failureKind, failureMessage });
|
|
return { closed: true, state: "closed-open-run", runStatus: next.status, terminalStatus, commandState: command.state, valuesPrinted: false };
|
|
}
|
|
|
|
function runnerJobAllowsRunClosure(phase: string | null): boolean {
|
|
return phase === "k8s:succeeded" || phase === "k8s:failed" || phase === "k8s:missing";
|
|
}
|
|
|
|
function terminalStatusFromCommand(command: CommandRecord): TerminalStatus {
|
|
if (command.state === "completed") return "completed";
|
|
if (command.state === "cancelled") return "cancelled";
|
|
return "failed";
|
|
}
|
|
|
|
function observedRunnerPhase(jobObject: K8sObject | null, pods: K8sObject[]): string {
|
|
if (!jobObject && pods.length === 0) return "k8s:missing";
|
|
const condition = jobCondition(jobObject);
|
|
if (condition === "Complete") return "k8s:succeeded";
|
|
if (condition === "Failed") return "k8s:failed";
|
|
const active = numberPath(jobObject, ["status", "active"]);
|
|
const succeeded = numberPath(jobObject, ["status", "succeeded"]);
|
|
const failed = numberPath(jobObject, ["status", "failed"]);
|
|
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 === "Pending")) return "k8s:pending";
|
|
return "k8s:unknown";
|
|
}
|
|
|
|
function categoryForPhase(phase: string): string {
|
|
if (phase === "k8s:succeeded") return "runner-job-k8s-succeeded";
|
|
if (phase === "k8s:failed") return "runner-job-k8s-failed";
|
|
if (phase === "k8s:missing") return "runner-job-k8s-missing";
|
|
if (phase === "k8s:running" || phase === "k8s:pending") return "runner-job-running";
|
|
return "runner-job-observed";
|
|
}
|
|
|
|
function terminalReportStateForPhase(phase: string): string {
|
|
if (phase === "k8s:succeeded" || phase === "k8s:failed") return "terminal-runner-report-missing";
|
|
if (phase === "k8s:missing") return "runner-resource-missing";
|
|
return "not-terminal";
|
|
}
|
|
|
|
function k8sSummary(jobObject: K8sObject | null, pods: K8sObject[]): JsonRecord {
|
|
const podPhases = pods.map((pod) => stringPath(pod, ["status", "phase"]) ?? "unknown");
|
|
return {
|
|
jobPresent: jobObject !== null,
|
|
jobUid: stringPath(jobObject, ["metadata", "uid"]),
|
|
condition: jobCondition(jobObject),
|
|
active: numberPath(jobObject, ["status", "active"]),
|
|
succeeded: numberPath(jobObject, ["status", "succeeded"]),
|
|
failed: numberPath(jobObject, ["status", "failed"]),
|
|
startTime: stringPath(jobObject, ["status", "startTime"]),
|
|
completionTime: stringPath(jobObject, ["status", "completionTime"]),
|
|
podCount: pods.length,
|
|
podPhases,
|
|
newestPodName: newestPodName(pods),
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function jobCondition(jobObject: K8sObject | null): string | null {
|
|
const conditions = jobObject?.status?.conditions;
|
|
if (!Array.isArray(conditions)) return null;
|
|
for (const condition of [...conditions].reverse()) {
|
|
if (typeof condition !== "object" || condition === null || Array.isArray(condition)) continue;
|
|
const record = condition as JsonRecord;
|
|
if ((record.type === "Complete" || record.type === "Failed") && record.status === "True") return record.type;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function newestPodName(pods: K8sObject[]): string | null {
|
|
const sorted = [...pods].sort((left, right) => (stringPath(right, ["metadata", "creationTimestamp"]) ?? "").localeCompare(stringPath(left, ["metadata", "creationTimestamp"]) ?? ""));
|
|
return stringPath(sorted[0] ?? null, ["metadata", "name"]);
|
|
}
|
|
|
|
async function getK8sObject(kubectlCommand: string, resource: string, namespace: string, name: string): Promise<K8sObject | null> {
|
|
const result = await kubectlRun(kubectlCommand, ["get", resource, name, "-n", namespace, "-o", "json"]);
|
|
if (result.code === 0) return parseJsonObject(result.stdout, `kubectl get ${resource}`) as K8sObject;
|
|
if (isNotFound(result.stderr)) return null;
|
|
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 }) });
|
|
}
|
|
|
|
async function getK8sList(kubectlCommand: string, resource: string, namespace: string, extraArgs: string[]): Promise<K8sObject[]> {
|
|
const result = await kubectlRun(kubectlCommand, ["get", resource, "-n", namespace, ...extraArgs, "-o", "json"]);
|
|
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}`) as K8sList;
|
|
return Array.isArray(parsed.items) ? parsed.items.filter((item) => typeof item === "object" && item !== null && !Array.isArray(item)) : [];
|
|
}
|
|
|
|
async function getRunnerJobLogs(kubectlCommand: string, namespace: string, jobName: string): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
|
return await kubectlRun(kubectlCommand, ["logs", `job/${jobName}`, "-n", namespace, "--tail=2000"]);
|
|
}
|
|
|
|
function findLatestTerminalOutbox(stdout: string, job: RunnerJobRecord): TerminalOutboxSearch {
|
|
let outbox: TerminalOutboxRecord | null = null;
|
|
let lineCount = 0;
|
|
let candidateCount = 0;
|
|
let parseFailedCount = 0;
|
|
for (const line of stdout.split(/\r?\n/u)) {
|
|
if (line.trim().length === 0) continue;
|
|
lineCount++;
|
|
if (!line.includes("terminal.outbox")) continue;
|
|
let parsed: unknown;
|
|
try {
|
|
parsed = JSON.parse(line);
|
|
} catch {
|
|
parseFailedCount++;
|
|
continue;
|
|
}
|
|
const candidate = terminalOutboxFromRecord(parsed);
|
|
if (!candidate) continue;
|
|
candidateCount++;
|
|
if (matchesTerminalOutbox(candidate, job)) outbox = candidate;
|
|
}
|
|
return { outbox, lineCount, candidateCount, parseFailedCount, stdoutHash: stableHash({ stdoutTail: stdout.slice(-20_000), stdoutLength: stdout.length }) };
|
|
}
|
|
|
|
function terminalOutboxFromRecord(value: unknown): TerminalOutboxRecord | null {
|
|
const record = recordValue(value);
|
|
if (!record || record.label !== "terminal.outbox") return null;
|
|
const outboxKey = stringValue(record.outboxKey);
|
|
const runId = stringValue(record.runId);
|
|
const commandId = stringValue(record.commandId);
|
|
const runnerId = stringValue(record.runnerId);
|
|
const attemptId = stringValue(record.attemptId);
|
|
const reportRecord = recordValue(record.report);
|
|
if (!outboxKey || !runId || !commandId || !runnerId || !attemptId || !reportRecord) return null;
|
|
const report = terminalReportFromRecord(reportRecord);
|
|
if (!report) return null;
|
|
const events = Array.isArray(record.events) ? record.events.map(eventFromRecord).filter((event): event is BackendEvent => event !== null) : [];
|
|
return {
|
|
outboxKey,
|
|
phase: stringValue(record.phase),
|
|
runId,
|
|
commandId,
|
|
runnerId,
|
|
runnerJobId: stringValue(record.runnerJobId),
|
|
attemptId,
|
|
report,
|
|
terminalRun: record.terminalRun === true,
|
|
events,
|
|
};
|
|
}
|
|
|
|
function terminalReportFromRecord(record: JsonRecord): CommandTerminalReport | null {
|
|
const terminalStatus = terminalStatusValue(record.terminalStatus);
|
|
const failureKind = failureKindValue(record.failureKind);
|
|
if (!terminalStatus || failureKind === undefined) return null;
|
|
const threadId = stringValue(record.threadId);
|
|
const turnId = stringValue(record.turnId);
|
|
return {
|
|
terminalStatus,
|
|
failureKind,
|
|
failureMessage: nullableString(record.failureMessage),
|
|
...(threadId ? { threadId } : {}),
|
|
...(turnId ? { turnId } : {}),
|
|
};
|
|
}
|
|
|
|
function eventFromRecord(value: unknown): BackendEvent | null {
|
|
const record = recordValue(value);
|
|
if (!record) return null;
|
|
const type = eventTypeValue(record.type);
|
|
const payload = recordValue(record.payload);
|
|
if (!type || !payload) return null;
|
|
return { type, payload };
|
|
}
|
|
|
|
function matchesTerminalOutbox(outbox: TerminalOutboxRecord, job: RunnerJobRecord): boolean {
|
|
if (outbox.runId !== job.runId || outbox.commandId !== job.commandId) return false;
|
|
if (outbox.runnerJobId && outbox.runnerJobId !== job.id) return false;
|
|
return true;
|
|
}
|
|
|
|
function eventContentKey(event: BackendEvent): string {
|
|
return stableHash({ type: event.type, payload: stripTerminalOutboxEventPayload(event.payload) });
|
|
}
|
|
|
|
function stripTerminalOutboxEventPayload(payload: JsonRecord): JsonRecord {
|
|
const next: JsonRecord = {};
|
|
for (const [key, value] of Object.entries(payload)) {
|
|
if (key === "terminalOutboxKey" || key === "terminalOutboxIndex") continue;
|
|
next[key] = value;
|
|
}
|
|
return next;
|
|
}
|
|
|
|
function recordValue(value: unknown): JsonRecord | null {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : null;
|
|
}
|
|
|
|
const eventTypes = new Set<string>(["backend_status", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"]);
|
|
const terminalStatuses = new Set<string>(["completed", "failed", "blocked", "cancelled"]);
|
|
const failureKinds = new Set<string>([
|
|
"auth-missing",
|
|
"auth-failed",
|
|
"schema-invalid",
|
|
"tenant-policy-denied",
|
|
"secret-unavailable",
|
|
"prompt-unavailable",
|
|
"prompt-too-large",
|
|
"required-skill-unavailable",
|
|
"skill-unavailable",
|
|
"runner-lease-conflict",
|
|
"backend-failed",
|
|
"backend-protocol-error",
|
|
"backend-spawn-failed",
|
|
"backend-json-parse-error",
|
|
"backend-response-invalid",
|
|
"thread-resume-failed",
|
|
"backend-timeout",
|
|
"provider-auth-failed",
|
|
"provider-rate-limited",
|
|
"provider-stream-disconnected",
|
|
"provider-http-error",
|
|
"provider-invalid-tool-call",
|
|
"provider-compact-unsupported",
|
|
"provider-unavailable",
|
|
"infra-failed",
|
|
"session-store-evicted",
|
|
"cancelled",
|
|
]);
|
|
|
|
function eventTypeValue(value: JsonValue | undefined): EventType | null {
|
|
return typeof value === "string" && eventTypes.has(value) ? value as EventType : null;
|
|
}
|
|
|
|
function terminalStatusValue(value: JsonValue | undefined): TerminalStatus | null {
|
|
return typeof value === "string" && terminalStatuses.has(value) ? value as TerminalStatus : null;
|
|
}
|
|
|
|
function failureKindValue(value: JsonValue | undefined): FailureKind | null | undefined {
|
|
if (value === null) return null;
|
|
if (typeof value === "string" && failureKinds.has(value)) return value as FailureKind;
|
|
return undefined;
|
|
}
|
|
|
|
function nullableString(value: JsonValue | undefined): string | null {
|
|
return typeof value === "string" ? value : null;
|
|
}
|
|
|
|
async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
|
const child = spawn(kubectlCommand, args, { stdio: ["ignore", "pipe", "pipe"] });
|
|
let stdout = "";
|
|
let stderr = "";
|
|
child.stdout.setEncoding("utf8");
|
|
child.stderr.setEncoding("utf8");
|
|
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
|
|
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
|
|
const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
|
|
child.on("error", reject);
|
|
child.on("close", (code, signal) => resolve({ code, signal }));
|
|
}).catch((error: unknown) => {
|
|
throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
|
|
});
|
|
return { ...result, stdout, stderr };
|
|
}
|
|
|
|
function parseJsonObject(text: string, label: 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 non-object JSON`, { httpStatus: 502, details: { stdoutHash: stableHash(text), valuesPrinted: false } });
|
|
}
|
|
|
|
function isNotFound(stderr: string): boolean {
|
|
return /notfound|not found|notfound/i.test(stderr.replace(/\s+/gu, ""));
|
|
}
|
|
|
|
function clampLimit(limit: number): number {
|
|
return Math.max(1, Math.min(Math.floor(limit), 500));
|
|
}
|
|
|
|
function stringValue(value: JsonValue | undefined): string | null {
|
|
return typeof value === "string" && value.length > 0 ? value : null;
|
|
}
|
|
|
|
function stringPath(value: unknown, path: string[]): string | null {
|
|
let current: unknown = value;
|
|
for (const key of path) {
|
|
if (typeof current !== "object" || current === null || Array.isArray(current)) return null;
|
|
current = (current as Record<string, unknown>)[key];
|
|
}
|
|
return typeof current === "string" ? current : null;
|
|
}
|
|
|
|
function numberPath(value: unknown, path: string[]): number | null {
|
|
let current: unknown = value;
|
|
for (const key of path) {
|
|
if (typeof current !== "object" || current === null || Array.isArray(current)) return null;
|
|
current = (current as Record<string, unknown>)[key];
|
|
}
|
|
return typeof current === "number" && Number.isFinite(current) ? current : null;
|
|
}
|