root
1224 lines
58 KiB
TypeScript
1224 lines
58 KiB
TypeScript
import { AgentRunError } from "../common/errors.js";
|
|
import { redactJson, redactText } from "../common/redaction.js";
|
|
import type { CommandRecord, JsonRecord, JsonValue, RunRecord } from "../common/types.js";
|
|
import { nowIso, stableHash } from "../common/validation.js";
|
|
import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
|
|
import type { AgentRunStore } from "./store.js";
|
|
import { createKubernetesClient, type KubernetesClient, type KubernetesResource, type KubernetesTransportResult } from "./kubernetes-transport.js";
|
|
|
|
export interface RunnerRetentionOptions {
|
|
namespace: string;
|
|
maxRunners: number;
|
|
cleanupOrder: "oldest-inactive-last-active-first";
|
|
activeHeartbeatMaxAgeMs: number;
|
|
stalePendingMaxAgeMs: number;
|
|
matchLabels: Record<string, string>;
|
|
jobNamePrefixes: string[];
|
|
ageBasedCleanup: {
|
|
enabled: boolean;
|
|
maxAgeHours: number | null;
|
|
};
|
|
kubectlCommand?: string;
|
|
signal?: AbortSignal;
|
|
}
|
|
|
|
export interface RunnerRetentionSummary extends JsonRecord {
|
|
enabled: boolean;
|
|
namespace: string;
|
|
maxRunners: number;
|
|
stalePendingMaxAgeMs: number;
|
|
incomingRunnerCount: number;
|
|
liveRunnerCountBefore: number;
|
|
liveRunnerCountAfter: number;
|
|
runnerJobCountBefore: number;
|
|
nonTerminalRunnerPodCountBefore: number;
|
|
orphanNonTerminalRunnerPodCountBefore: number;
|
|
inactiveCandidateCount: number;
|
|
protectedActiveRunnerCount: number;
|
|
protectedActiveRunners: JsonRecord[];
|
|
selectedRunnerCount: number;
|
|
deletedRunnerJobCount: number;
|
|
deletedRunnerPodCount: number;
|
|
deletedAssociatedResourceCount: number;
|
|
repairedTerminalClaimCount: number;
|
|
activeRunRisk: boolean;
|
|
reason: string;
|
|
valuesPrinted: false;
|
|
}
|
|
|
|
export interface RunnerStalePendingReconcileSummary extends JsonRecord {
|
|
enabled: true;
|
|
action: "runner-stale-pending-reconcile";
|
|
observedAt: string;
|
|
namespace: string;
|
|
limit: number;
|
|
maxRunners: number;
|
|
stalePendingMaxAgeMs: number;
|
|
observedRunnerCount: number;
|
|
liveRunnerCountBefore: number;
|
|
liveRunnerCountAfter: number | null;
|
|
stalePendingCandidateCount: number;
|
|
protectedActiveRunnerCount: number;
|
|
selectedRunnerCount: number;
|
|
terminalizedRunCount: number;
|
|
terminalizedCommandCount: number;
|
|
deletedRunnerJobCount: number;
|
|
deletedRunnerPodCount: number;
|
|
deletedAssociatedResourceCount: number;
|
|
failedRunnerCount: number;
|
|
reason: string;
|
|
valuesPrinted: false;
|
|
}
|
|
|
|
type CandidateKind = "idle" | "terminal" | "inactive" | "stale-pending";
|
|
type RepairKind = "none" | "release-terminal-run-claim" | "release-terminal-command-claim" | "terminalize-stale-pending";
|
|
|
|
interface RunnerPodObservation extends JsonRecord {
|
|
name: string;
|
|
uid: string | null;
|
|
resourceVersion: string | null;
|
|
phase: string | null;
|
|
createdAt: string | null;
|
|
ageMs: number | null;
|
|
stale: boolean;
|
|
podStartTime: string | null;
|
|
ready: boolean;
|
|
runnerContainerState: "waiting" | "running" | "terminated" | "unknown";
|
|
runnerContainerLastState: "running" | "terminated" | "none";
|
|
runnerContainerStarted: boolean;
|
|
runnerContainerReady: boolean;
|
|
runnerContainerRestartCount: number;
|
|
waiting: JsonRecord[];
|
|
waitingEvidenceSources: string[];
|
|
eventFactsAvailable: boolean;
|
|
factsAvailable: boolean;
|
|
valuesPrinted: false;
|
|
}
|
|
|
|
interface K8sObject {
|
|
kind?: string;
|
|
metadata?: {
|
|
name?: string;
|
|
namespace?: string;
|
|
uid?: string;
|
|
resourceVersion?: string;
|
|
creationTimestamp?: string;
|
|
labels?: Record<string, string>;
|
|
annotations?: Record<string, string>;
|
|
ownerReferences?: Array<{ kind?: string; name?: string }>;
|
|
};
|
|
status?: JsonRecord;
|
|
involvedObject?: { kind?: string; name?: string };
|
|
reason?: string;
|
|
message?: string;
|
|
type?: string;
|
|
count?: number;
|
|
firstTimestamp?: string;
|
|
lastTimestamp?: string;
|
|
eventTime?: string;
|
|
series?: JsonRecord;
|
|
}
|
|
|
|
interface RunnerResourceEntry {
|
|
resourceKind: "job" | "pod";
|
|
name: string;
|
|
jobName: string;
|
|
resourceUid: string | null;
|
|
resourceVersion: string | null;
|
|
runId: string | null;
|
|
commandId: string | null;
|
|
runnerId: string | null;
|
|
createdAtMs: number;
|
|
terminal: boolean;
|
|
podPhase: string | null;
|
|
podObservations: RunnerPodObservation[];
|
|
protectedActive: boolean;
|
|
protectedReason: string | null;
|
|
candidateKind: CandidateKind;
|
|
classificationReason: string;
|
|
repairKind: RepairKind;
|
|
lastActiveAtMs: number;
|
|
dbFactsFingerprint: string | null;
|
|
}
|
|
|
|
interface Snapshot {
|
|
jobs: RunnerResourceEntry[];
|
|
pods: RunnerResourceEntry[];
|
|
liveRunnerCount: number;
|
|
runnerJobCount: number;
|
|
nonTerminalRunnerPodCount: number;
|
|
orphanNonTerminalRunnerPodCount: number;
|
|
}
|
|
|
|
export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRunStore; options: RunnerRetentionOptions; incomingRunnerCount?: number }): Promise<RunnerRetentionSummary> {
|
|
const incomingRunnerCount = input.incomingRunnerCount ?? 1;
|
|
const before = await runnerSnapshot(input.store, input.options);
|
|
const targetBeforeCreate = Math.max(0, input.options.maxRunners - incomingRunnerCount);
|
|
const requiredDeleteCount = Math.max(0, before.liveRunnerCount - targetBeforeCreate);
|
|
const candidates = [...before.jobs, ...before.pods]
|
|
.filter((item) => !item.protectedActive)
|
|
.sort(compareCandidates);
|
|
const selected = candidates.slice(0, requiredDeleteCount);
|
|
const protectedActiveRunners = [...before.jobs, ...before.pods].filter((item) => item.protectedActive);
|
|
const protectedActiveRunnerCount = protectedActiveRunners.length;
|
|
const inactiveCandidates = boundedResourceSummaries(candidates);
|
|
const protectedSummaries = boundedResourceSummaries(protectedActiveRunners);
|
|
if (requiredDeleteCount > 0 && selected.length < requiredDeleteCount) {
|
|
throw new AgentRunError("infra-failed", "runner retention cannot safely make room for a new runner", {
|
|
httpStatus: 409,
|
|
details: {
|
|
reason: "runner-retention-no-safe-candidate",
|
|
namespace: input.options.namespace,
|
|
maxRunners: input.options.maxRunners,
|
|
stalePendingMaxAgeMs: input.options.stalePendingMaxAgeMs,
|
|
incomingRunnerCount,
|
|
liveRunnerCountBefore: before.liveRunnerCount,
|
|
requiredDeleteCount,
|
|
inactiveCandidateCount: candidates.length,
|
|
inactiveCandidates: inactiveCandidates.items,
|
|
inactiveCandidateOmittedCount: inactiveCandidates.omittedCount,
|
|
protectedActiveRunnerCount,
|
|
protectedActiveRunners: protectedSummaries.items,
|
|
protectedActiveRunnerOmittedCount: protectedSummaries.omittedCount,
|
|
activeRunRisk: protectedActiveRunnerCount > 0,
|
|
cleanupPolicy: cleanupPolicySummary(),
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
}
|
|
|
|
let deletedRunnerJobCount = 0;
|
|
let deletedRunnerPodCount = 0;
|
|
let deletedAssociatedResourceCount = 0;
|
|
let repairedTerminalClaimCount = 0;
|
|
for (const item of selected) {
|
|
const retired = await retireSelectedRunner(input.store, input.options, item, `runner retention retired stale Kubernetes ${item.resourceKind}/${item.name}`);
|
|
deletedRunnerJobCount += retired.deletedRunnerJobCount;
|
|
deletedRunnerPodCount += retired.deletedRunnerPodCount;
|
|
repairedTerminalClaimCount += retired.repairedTerminalClaimCount;
|
|
if (item.resourceKind === "job") deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item);
|
|
}
|
|
const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before;
|
|
if (after.liveRunnerCount > targetBeforeCreate) {
|
|
throw new AgentRunError("infra-failed", "runner retention did not make enough room for a new runner", {
|
|
httpStatus: 409,
|
|
details: {
|
|
reason: "runner-retention-postcondition-failed",
|
|
namespace: input.options.namespace,
|
|
maxRunners: input.options.maxRunners,
|
|
incomingRunnerCount,
|
|
targetBeforeCreate,
|
|
liveRunnerCountBefore: before.liveRunnerCount,
|
|
liveRunnerCountAfter: after.liveRunnerCount,
|
|
selectedRunnerCount: selected.length,
|
|
deletedRunnerJobCount,
|
|
deletedRunnerPodCount,
|
|
deletedAssociatedResourceCount,
|
|
repairedTerminalClaimCount,
|
|
selected: selected.map(resourceSummary),
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
}
|
|
return {
|
|
enabled: true,
|
|
namespace: input.options.namespace,
|
|
maxRunners: input.options.maxRunners,
|
|
stalePendingMaxAgeMs: input.options.stalePendingMaxAgeMs,
|
|
incomingRunnerCount,
|
|
liveRunnerCountBefore: before.liveRunnerCount,
|
|
liveRunnerCountAfter: after.liveRunnerCount,
|
|
runnerJobCountBefore: before.runnerJobCount,
|
|
nonTerminalRunnerPodCountBefore: before.nonTerminalRunnerPodCount,
|
|
orphanNonTerminalRunnerPodCountBefore: before.orphanNonTerminalRunnerPodCount,
|
|
inactiveCandidateCount: candidates.length,
|
|
inactiveCandidates: inactiveCandidates.items,
|
|
inactiveCandidateOmittedCount: inactiveCandidates.omittedCount,
|
|
protectedActiveRunnerCount,
|
|
protectedActiveRunners: protectedSummaries.items,
|
|
protectedActiveRunnerOmittedCount: protectedSummaries.omittedCount,
|
|
selectedRunnerCount: selected.length,
|
|
deletedRunnerJobCount,
|
|
deletedRunnerPodCount,
|
|
deletedAssociatedResourceCount,
|
|
repairedTerminalClaimCount,
|
|
activeRunRisk: protectedActiveRunnerCount > 0,
|
|
reason: requiredDeleteCount === 0 ? "within-limit" : "pre-create-retention",
|
|
cleanupPolicy: cleanupPolicySummary(),
|
|
selected: selected.map(resourceSummary),
|
|
valuesPrinted: false,
|
|
} satisfies RunnerRetentionSummary;
|
|
}
|
|
|
|
export async function reconcileStalePendingRunners(input: { store: AgentRunStore; options: RunnerRetentionOptions; limit: number }): Promise<RunnerStalePendingReconcileSummary> {
|
|
const limit = clampReconcileLimit(input.limit);
|
|
return await input.store.withRunnerCreateFence(input.options.namespace, async () => {
|
|
const observedAt = nowIso();
|
|
const before = await runnerSnapshot(input.store, input.options);
|
|
const resources = [...before.jobs, ...before.pods];
|
|
const stalePendingCandidates = resources
|
|
.filter((item) => !item.protectedActive && item.candidateKind === "stale-pending")
|
|
.sort(compareCandidates);
|
|
const protectedActiveRunners = resources.filter((item) => item.protectedActive);
|
|
const selected = stalePendingCandidates.slice(0, limit);
|
|
const outcomes: JsonRecord[] = [];
|
|
let terminalizedRunCount = 0;
|
|
let terminalizedCommandCount = 0;
|
|
let deletedRunnerJobCount = 0;
|
|
let deletedRunnerPodCount = 0;
|
|
let deletedAssociatedResourceCount = 0;
|
|
let failedRunnerCount = 0;
|
|
|
|
for (const item of selected) {
|
|
let retired: RetiredRunnerResource | null = null;
|
|
try {
|
|
retired = await retireSelectedRunner(input.store, input.options, item, `runner reconciler retired stale-pending Kubernetes ${item.resourceKind}/${item.name}`);
|
|
terminalizedRunCount += item.candidateKind === "stale-pending" ? 1 : 0;
|
|
terminalizedCommandCount += item.candidateKind === "stale-pending" ? 1 : 0;
|
|
deletedRunnerJobCount += retired.deletedRunnerJobCount;
|
|
deletedRunnerPodCount += retired.deletedRunnerPodCount;
|
|
const deletedAssociatedForRunner = item.resourceKind === "job" ? await deleteAssociatedResources(input.options, item) : 0;
|
|
deletedAssociatedResourceCount += deletedAssociatedForRunner;
|
|
outcomes.push({ ...resourceIdentitySummary(item), state: "retired", terminalized: item.candidateKind === "stale-pending", ...retired, deletedAssociatedResourceCount: deletedAssociatedForRunner, valuesPrinted: false });
|
|
} catch (error) {
|
|
failedRunnerCount++;
|
|
outcomes.push({ ...resourceIdentitySummary(item), state: retired ? "retired-associated-cleanup-failed" : "failed-closed", coreRetired: retired !== null, ...(retired ?? {}), error: retentionErrorSummary(error), valuesPrinted: false });
|
|
}
|
|
}
|
|
|
|
let after: Snapshot | null = before;
|
|
let postObservationError: JsonRecord | null = null;
|
|
if (selected.length > 0) {
|
|
try {
|
|
after = await runnerSnapshot(input.store, input.options);
|
|
} catch (error) {
|
|
after = null;
|
|
postObservationError = retentionErrorSummary(error);
|
|
}
|
|
}
|
|
const candidateSummaries = boundedResourceSummaries(stalePendingCandidates);
|
|
const protectedSummaries = boundedResourceSummaries(protectedActiveRunners);
|
|
const selectedSummaries = boundedResourceSummaries(selected);
|
|
const boundedOutcomes = boundedJsonRecords(outcomes);
|
|
return {
|
|
enabled: true,
|
|
action: "runner-stale-pending-reconcile",
|
|
observedAt,
|
|
namespace: input.options.namespace,
|
|
limit,
|
|
maxRunners: input.options.maxRunners,
|
|
stalePendingMaxAgeMs: input.options.stalePendingMaxAgeMs,
|
|
observedRunnerCount: resources.length,
|
|
liveRunnerCountBefore: before.liveRunnerCount,
|
|
liveRunnerCountAfter: after?.liveRunnerCount ?? null,
|
|
runnerJobCountBefore: before.runnerJobCount,
|
|
nonTerminalRunnerPodCountBefore: before.nonTerminalRunnerPodCount,
|
|
orphanNonTerminalRunnerPodCountBefore: before.orphanNonTerminalRunnerPodCount,
|
|
inactiveCandidateCount: resources.filter((item) => !item.protectedActive).length,
|
|
stalePendingCandidateCount: stalePendingCandidates.length,
|
|
stalePendingCandidates: candidateSummaries.items,
|
|
stalePendingCandidateOmittedCount: candidateSummaries.omittedCount,
|
|
protectedActiveRunnerCount: protectedActiveRunners.length,
|
|
protectedActiveRunners: protectedSummaries.items,
|
|
protectedActiveRunnerOmittedCount: protectedSummaries.omittedCount,
|
|
protectedReasonCounts: reasonCounts(protectedActiveRunners.map((item) => item.protectedReason ?? "unknown")),
|
|
selectedRunnerCount: selected.length,
|
|
selectedRunners: selectedSummaries.items,
|
|
selectedRunnerOmittedCount: selectedSummaries.omittedCount,
|
|
terminalizedRunCount,
|
|
terminalizedCommandCount,
|
|
deletedRunnerJobCount,
|
|
deletedRunnerPodCount,
|
|
deletedAssociatedResourceCount,
|
|
failedRunnerCount,
|
|
outcomes: boundedOutcomes.items,
|
|
outcomeOmittedCount: boundedOutcomes.omittedCount,
|
|
state: postObservationError || failedRunnerCount > 0 ? "partial" : "completed",
|
|
postObservationState: postObservationError ? "failed" : "completed",
|
|
postObservationError,
|
|
activeRunRisk: protectedActiveRunners.length > 0,
|
|
reason: postObservationError ? "stale-pending-post-observation-failed" : selected.length === 0 ? "no-stale-pending-candidate" : failedRunnerCount > 0 ? "stale-pending-reconcile-partial" : "stale-pending-reconciled",
|
|
cleanupPolicy: cleanupPolicySummary(),
|
|
valuesPrinted: false,
|
|
} satisfies RunnerStalePendingReconcileSummary;
|
|
});
|
|
}
|
|
|
|
async function runnerSnapshot(store: AgentRunStore, options: RunnerRetentionOptions): Promise<Snapshot> {
|
|
const observed = await Promise.allSettled([
|
|
kubernetesGetList(options, "jobs", labelSelector(options.matchLabels)),
|
|
kubernetesGetList(options, "pods", labelSelector(options.matchLabels)),
|
|
]);
|
|
const failure = observed.find((item): item is PromiseRejectedResult => item.status === "rejected");
|
|
if (failure) throw failure.reason;
|
|
const [jobObjects, podObjects] = observed.map((item) => (item as PromiseFulfilledResult<K8sObject[]>).value);
|
|
const matchingPods = podObjects.filter((item) => {
|
|
const name = objectName(item);
|
|
if (!name) return false;
|
|
return matchesJobPrefix(podJobName(item) ?? name, options.jobNamePrefixes);
|
|
});
|
|
const podsByJob = new Map<string, K8sObject[]>();
|
|
const eventsByPod = new Map<string, K8sObject[]>();
|
|
let eventFactsAvailable = true;
|
|
let eventObjects: K8sObject[] = [];
|
|
if (matchingPods.length > 0) {
|
|
try {
|
|
eventObjects = await kubernetesGetPodEvents(options);
|
|
} catch (error) {
|
|
if (options.signal?.aborted) throw error;
|
|
eventFactsAvailable = false;
|
|
}
|
|
}
|
|
for (const item of eventObjects) {
|
|
const podName = podEventName(item);
|
|
if (!podName) continue;
|
|
const current = eventsByPod.get(podName) ?? [];
|
|
current.push(item);
|
|
eventsByPod.set(podName, current);
|
|
}
|
|
for (const item of matchingPods) {
|
|
const name = objectName(item);
|
|
if (!name) continue;
|
|
const jobName = podJobName(item) ?? name;
|
|
const current = podsByJob.get(jobName) ?? [];
|
|
current.push(item);
|
|
podsByJob.set(jobName, current);
|
|
}
|
|
const jobs: RunnerResourceEntry[] = [];
|
|
for (const item of jobObjects) {
|
|
const name = objectName(item);
|
|
if (!name || !matchesJobPrefix(name, options.jobNamePrefixes)) continue;
|
|
jobs.push(await jobEntry(store, item, options, podsByJob.get(name) ?? [], eventsByPod, eventFactsAvailable));
|
|
}
|
|
const jobNames = new Set(jobs.map((item) => item.jobName));
|
|
const pods: RunnerResourceEntry[] = [];
|
|
let nonTerminalRunnerPodCount = 0;
|
|
let orphanNonTerminalRunnerPodCount = 0;
|
|
for (const item of matchingPods) {
|
|
const name = objectName(item);
|
|
if (!name) continue;
|
|
const jobName = podJobName(item) ?? name;
|
|
if (!matchesJobPrefix(jobName, options.jobNamePrefixes)) continue;
|
|
const terminal = podTerminal(item);
|
|
if (!terminal) nonTerminalRunnerPodCount++;
|
|
if (jobNames.has(jobName)) continue;
|
|
if (!terminal) orphanNonTerminalRunnerPodCount++;
|
|
pods.push(await podEntry(store, item, options, jobName, eventsByPod.get(name) ?? [], eventFactsAvailable));
|
|
}
|
|
return {
|
|
jobs,
|
|
pods,
|
|
liveRunnerCount: jobs.length + pods.filter((item) => !item.terminal).length,
|
|
runnerJobCount: jobs.length,
|
|
nonTerminalRunnerPodCount,
|
|
orphanNonTerminalRunnerPodCount,
|
|
};
|
|
}
|
|
|
|
async function jobEntry(store: AgentRunStore, item: K8sObject, options: RunnerRetentionOptions, pods: K8sObject[], eventsByPod: Map<string, K8sObject[]>, eventFactsAvailable: boolean): Promise<RunnerResourceEntry> {
|
|
const name = objectName(item) ?? "unknown";
|
|
const annotations = item.metadata?.annotations ?? {};
|
|
const podObservations = pods.map((pod) => observePod(pod, options.stalePendingMaxAgeMs, eventsByPod.get(objectName(pod) ?? "") ?? [], eventFactsAvailable));
|
|
const podIdentityMatch = pods.every((pod) => resourceIdentityMatches(annotations, pod.metadata?.annotations ?? {}));
|
|
const activity = await activityFor(store, {
|
|
runId: annotations["agentrun.pikastech.local/run-id"] ?? null,
|
|
commandId: annotations["agentrun.pikastech.local/command-id"] ?? null,
|
|
runnerId: annotations["agentrun.pikastech.local/runner-id"] ?? null,
|
|
terminal: jobTerminal(item),
|
|
createdAtMs: objectCreatedAtMs(item),
|
|
activeHeartbeatMaxAgeMs: options.activeHeartbeatMaxAgeMs,
|
|
stalePendingMaxAgeMs: options.stalePendingMaxAgeMs,
|
|
podObservations,
|
|
podIdentityMatch,
|
|
});
|
|
return {
|
|
resourceKind: "job",
|
|
name,
|
|
jobName: name,
|
|
resourceUid: item.metadata?.uid ?? null,
|
|
resourceVersion: item.metadata?.resourceVersion ?? null,
|
|
createdAtMs: objectCreatedAtMs(item),
|
|
terminal: jobTerminal(item),
|
|
podPhase: null,
|
|
podObservations,
|
|
...activity,
|
|
};
|
|
}
|
|
|
|
async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRetentionOptions, jobName: string, events: K8sObject[], eventFactsAvailable: boolean): Promise<RunnerResourceEntry> {
|
|
const name = objectName(item) ?? "unknown";
|
|
const annotations = item.metadata?.annotations ?? {};
|
|
const podObservations = [observePod(item, options.stalePendingMaxAgeMs, events, eventFactsAvailable)];
|
|
const activity = await activityFor(store, {
|
|
runId: annotations["agentrun.pikastech.local/run-id"] ?? null,
|
|
commandId: annotations["agentrun.pikastech.local/command-id"] ?? null,
|
|
runnerId: annotations["agentrun.pikastech.local/runner-id"] ?? null,
|
|
terminal: podTerminal(item),
|
|
createdAtMs: objectCreatedAtMs(item),
|
|
activeHeartbeatMaxAgeMs: options.activeHeartbeatMaxAgeMs,
|
|
stalePendingMaxAgeMs: options.stalePendingMaxAgeMs,
|
|
podObservations,
|
|
podIdentityMatch: true,
|
|
});
|
|
return {
|
|
resourceKind: "pod",
|
|
name,
|
|
jobName,
|
|
resourceUid: item.metadata?.uid ?? null,
|
|
resourceVersion: item.metadata?.resourceVersion ?? null,
|
|
createdAtMs: objectCreatedAtMs(item),
|
|
terminal: podTerminal(item),
|
|
podPhase: stringPath(item, ["status", "phase"]),
|
|
podObservations,
|
|
...activity,
|
|
};
|
|
}
|
|
|
|
type ActivityFields = Pick<RunnerResourceEntry, "runId" | "commandId" | "runnerId" | "protectedActive" | "protectedReason" | "candidateKind" | "classificationReason" | "repairKind" | "lastActiveAtMs" | "dbFactsFingerprint">;
|
|
|
|
async function activityFor(store: AgentRunStore, input: {
|
|
runId: string | null;
|
|
commandId: string | null;
|
|
runnerId: string | null;
|
|
terminal: boolean;
|
|
createdAtMs: number;
|
|
activeHeartbeatMaxAgeMs: number;
|
|
stalePendingMaxAgeMs: number;
|
|
podObservations: RunnerPodObservation[];
|
|
podIdentityMatch: boolean;
|
|
}): Promise<ActivityFields> {
|
|
if (!input.runId || !input.commandId || !input.runnerId) return protectedActivity(input, "identity-facts-unavailable", input.createdAtMs, null);
|
|
if (!input.podIdentityMatch) return protectedActivity(input, "pod-identity-mismatch", input.createdAtMs, null);
|
|
|
|
let run: RunRecord;
|
|
let command: CommandRecord;
|
|
try {
|
|
[run, command] = await Promise.all([store.getRun(input.runId), store.getCommand(input.commandId)]);
|
|
} catch {
|
|
return protectedActivity(input, "db-facts-unavailable", input.createdAtMs, null);
|
|
}
|
|
|
|
const lastActiveAtMs = Math.max(input.createdAtMs, isoMs(run.updatedAt), isoMs(command.updatedAt), isoMs(run.leaseExpiresAt));
|
|
if (command.runId !== run.id) return protectedActivity(input, "db-identity-mismatch", lastActiveAtMs, null);
|
|
let commands: CommandRecord[];
|
|
let events: import("../common/types.js").RunEvent[];
|
|
try {
|
|
[commands, events] = await Promise.all([store.listCommands(run.id, 0, 100), store.listEvents(run.id, 0, 500)]);
|
|
} catch {
|
|
return protectedActivity(input, "db-progress-facts-unavailable", lastActiveAtMs, null);
|
|
}
|
|
const dbFactsFingerprint = databaseFactsFingerprint(run, command, commands, events);
|
|
const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs));
|
|
if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs, dbFactsFingerprint);
|
|
const activePodReason = activePodProtectionReason(input.podObservations);
|
|
if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs, dbFactsFingerprint);
|
|
if (isTerminalRunStatus(run.status)) {
|
|
const claimDrift = run.claimedBy !== null || run.leaseExpiresAt !== null;
|
|
return candidateActivity(input, "terminal", claimDrift ? "terminal-run-stale-claim" : "terminal-run", lastActiveAtMs, dbFactsFingerprint, claimDrift ? "release-terminal-run-claim" : "none");
|
|
}
|
|
if (isTerminalCommandState(command.state)) {
|
|
const claimDrift = input.terminal && (run.claimedBy !== null || run.leaseExpiresAt !== null);
|
|
return candidateActivity(
|
|
input,
|
|
input.terminal ? "terminal" : "idle",
|
|
claimDrift ? "terminal-command-stale-claim" : input.terminal ? "terminal-runner-resource" : "terminal-command-idle-resource",
|
|
lastActiveAtMs,
|
|
dbFactsFingerprint,
|
|
claimDrift ? "release-terminal-command-claim" : "none",
|
|
);
|
|
}
|
|
if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs, dbFactsFingerprint);
|
|
|
|
const podBlockReason = stalePendingPodBlockReason(input.podObservations);
|
|
if (podBlockReason) return protectedActivity(input, podBlockReason, lastActiveAtMs, dbFactsFingerprint);
|
|
if (!hasNoBusinessProgress(run, command, commands, events, input.stalePendingMaxAgeMs)) {
|
|
return protectedActivity(input, "business-progress-observed-or-unverifiable", lastActiveAtMs, dbFactsFingerprint);
|
|
}
|
|
return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint, "terminalize-stale-pending");
|
|
}
|
|
|
|
interface RetiredRunnerResource extends JsonRecord {
|
|
deletedRunnerJobCount: number;
|
|
deletedRunnerPodCount: number;
|
|
repairedTerminalClaimCount: number;
|
|
}
|
|
|
|
async function retireSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, item: RunnerResourceEntry, reason: string): Promise<RetiredRunnerResource> {
|
|
await revalidateSelectedRunner(store, options, item);
|
|
if (!item.runId || !item.commandId || !item.dbFactsFingerprint) {
|
|
throw new AgentRunError("infra-failed", "runner retention selected resource has no fenceable database identity", {
|
|
httpStatus: 409,
|
|
details: { reason: "runner-retention-db-fence-unavailable", namespace: options.namespace, resourceKind: item.resourceKind, name: item.name, valuesPrinted: false },
|
|
});
|
|
}
|
|
await store.withRunnerRetentionFence({
|
|
runId: item.runId,
|
|
commandId: item.commandId,
|
|
terminalizeStalePending: item.candidateKind === "stale-pending",
|
|
releaseRunnerClaim: item.repairKind === "release-terminal-run-claim" || item.repairKind === "release-terminal-command-claim",
|
|
reason,
|
|
}, async (facts) => {
|
|
const currentDbFingerprint = databaseFactsFingerprint(facts.run, facts.command, facts.commands, facts.events);
|
|
if (currentDbFingerprint !== item.dbFactsFingerprint) throw databaseFenceRejected(options, item, currentDbFingerprint, facts.run);
|
|
await deleteRunnerResourceCas(options, item);
|
|
});
|
|
if (item.resourceKind === "job") {
|
|
return { deletedRunnerJobCount: 1, deletedRunnerPodCount: 0, repairedTerminalClaimCount: isClaimRepair(item.repairKind) ? 1 : 0 };
|
|
}
|
|
return { deletedRunnerJobCount: 0, deletedRunnerPodCount: 1, repairedTerminalClaimCount: isClaimRepair(item.repairKind) ? 1 : 0 };
|
|
}
|
|
|
|
function isClaimRepair(repairKind: RepairKind): boolean {
|
|
return repairKind === "release-terminal-run-claim" || repairKind === "release-terminal-command-claim";
|
|
}
|
|
|
|
async function revalidateSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, selected: RunnerResourceEntry): Promise<void> {
|
|
const snapshot = await runnerSnapshot(store, options);
|
|
const resources = selected.resourceKind === "job" ? snapshot.jobs : snapshot.pods;
|
|
const current = resources.find((item) => item.name === selected.name) ?? null;
|
|
const selectedFingerprint = runnerSafetyFingerprint(selected);
|
|
const currentFingerprint = current ? runnerSafetyFingerprint(current) : null;
|
|
if (current && !current.protectedActive && currentFingerprint === selectedFingerprint) return;
|
|
|
|
throw new AgentRunError("infra-failed", "runner retention selected resource changed before deletion", {
|
|
httpStatus: 409,
|
|
details: {
|
|
reason: "runner-retention-selected-runner-changed",
|
|
changeReason: current === null ? "resource-missing" : current.protectedActive ? "protected-active" : "safety-fingerprint-changed",
|
|
namespace: options.namespace,
|
|
resourceKind: selected.resourceKind,
|
|
name: selected.name,
|
|
selectedFingerprint,
|
|
currentFingerprint,
|
|
selected: resourceSummary(selected),
|
|
current: current ? resourceSummary(current) : null,
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
}
|
|
|
|
function databaseFenceRejected(options: RunnerRetentionOptions, item: RunnerResourceEntry, currentDbFingerprint: string, run: RunRecord): AgentRunError {
|
|
return new AgentRunError("infra-failed", "runner retention database facts changed before atomic deletion", {
|
|
httpStatus: 409,
|
|
details: {
|
|
reason: "runner-retention-db-fence-rejected",
|
|
changeReason: run.claimedBy || run.status === "claimed" || run.status === "running" ? "run-became-active" : "database-fingerprint-changed",
|
|
namespace: options.namespace,
|
|
resourceKind: item.resourceKind,
|
|
name: item.name,
|
|
selectedDbFingerprint: item.dbFactsFingerprint,
|
|
currentDbFingerprint,
|
|
runStatus: run.status,
|
|
claimedBy: run.claimedBy,
|
|
leaseExpiresAt: run.leaseExpiresAt,
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
}
|
|
|
|
async function deleteRunnerResourceCas(options: RunnerRetentionOptions, item: RunnerResourceEntry): Promise<void> {
|
|
if (item.resourceKind === "job") {
|
|
for (const pod of [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name))) {
|
|
await deleteKubernetesObjectCas(options, item, "pod", pod.name, pod.uid, pod.resourceVersion);
|
|
}
|
|
const currentJob = await kubernetesGetObject(options, "job", item.name);
|
|
const expectedIdentity = {
|
|
"agentrun.pikastech.local/run-id": item.runId ?? "",
|
|
"agentrun.pikastech.local/command-id": item.commandId ?? "",
|
|
"agentrun.pikastech.local/runner-id": item.runnerId ?? "",
|
|
};
|
|
if (currentJob.metadata?.uid !== item.resourceUid || !resourceIdentityMatches(expectedIdentity, currentJob.metadata?.annotations ?? {})) {
|
|
throw new AgentRunError("infra-failed", "runner retention Job identity changed after Pod CAS deletion", {
|
|
httpStatus: 409,
|
|
details: { reason: "runner-retention-k8s-cas-rejected", changeReason: "job-identity-changed", namespace: options.namespace, resourceKind: "job", name: item.name, expectedUid: item.resourceUid, currentUid: currentJob.metadata?.uid ?? null, valuesPrinted: false },
|
|
});
|
|
}
|
|
await deleteKubernetesObjectCas(options, item, "job", item.name, currentJob.metadata?.uid ?? null, currentJob.metadata?.resourceVersion ?? null);
|
|
return;
|
|
}
|
|
await deleteKubernetesObjectCas(options, item, item.resourceKind, item.name, item.resourceUid, item.resourceVersion);
|
|
}
|
|
|
|
async function deleteKubernetesObjectCas(options: RunnerRetentionOptions, selected: RunnerResourceEntry, resourceKind: "job" | "pod", name: string, uid: string | null, resourceVersion: string | null): Promise<void> {
|
|
if (!uid || !resourceVersion) {
|
|
throw new AgentRunError("infra-failed", "runner retention cannot bind Kubernetes delete preconditions", {
|
|
httpStatus: 409,
|
|
details: { reason: "runner-retention-k8s-cas-unavailable", namespace: options.namespace, selectedResourceKind: selected.resourceKind, selectedName: selected.name, resourceKind, name, uidPresent: Boolean(uid), resourceVersionPresent: Boolean(resourceVersion), valuesPrinted: false },
|
|
});
|
|
}
|
|
const result = await retentionKubernetes(options).delete(resourceKind, name, { propagationPolicy: "Background", preconditions: { uid, resourceVersion } });
|
|
if (result.code === 0) return;
|
|
const conflict = result.kubernetesReason === "Conflict" || /\b(?:Conflict|PreconditionFailed|resourceVersion|UID)\b/iu.test(`${result.stderr}\n${result.stdout}`);
|
|
throw new AgentRunError("infra-failed", `Kubernetes CAS delete ${resourceKind} failed with code ${result.code}`, {
|
|
httpStatus: conflict ? 409 : 502,
|
|
details: redactJson({
|
|
reason: "runner-retention-k8s-cas-rejected",
|
|
changeReason: conflict ? "object-identity-or-version-changed" : "delete-request-failed",
|
|
namespace: options.namespace,
|
|
selectedResourceKind: selected.resourceKind,
|
|
selectedName: selected.name,
|
|
resourceKind,
|
|
name,
|
|
expectedUid: uid,
|
|
expectedResourceVersion: resourceVersion,
|
|
exitCode: result.code,
|
|
signal: result.signal,
|
|
transport: result.transport,
|
|
statusCode: result.statusCode,
|
|
kubernetesReason: result.kubernetesReason,
|
|
stderr: redactText(result.stderr.slice(-2000)),
|
|
stdoutHash: result.stdout ? stableHash(result.stdout) : null,
|
|
valuesPrinted: false,
|
|
}),
|
|
});
|
|
}
|
|
|
|
async function kubernetesGetList(options: RunnerRetentionOptions, resource: string, selector: string): Promise<K8sObject[]> {
|
|
const result = await retentionKubernetes(options).list(retentionResource(resource), selector ? { labelSelector: selector } : {});
|
|
if (result.code !== 0) throw new AgentRunError("infra-failed", `Kubernetes get ${resource} failed`, { httpStatus: 502, details: retentionFailureDetails(result, { reason: "runner-kubernetes-observation-failed", namespace: options.namespace, resource }) });
|
|
const parsed = parseJsonObject(result.stdout, `Kubernetes 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) : [];
|
|
}
|
|
|
|
async function kubernetesGetObject(options: RunnerRetentionOptions, resource: "job" | "pod", name: string): Promise<K8sObject> {
|
|
const result = await retentionKubernetes(options).get(resource, name);
|
|
if (result.code !== 0) throw new AgentRunError("infra-failed", `Kubernetes get ${resource}/${name} failed`, { httpStatus: 409, details: retentionFailureDetails(result, { reason: "runner-retention-k8s-cas-rejected", changeReason: "resource-missing-before-cas", namespace: options.namespace, resourceKind: resource, name }) });
|
|
return parseJsonObject(result.stdout, `Kubernetes get ${resource}/${name}`, "runner-retention-k8s-cas-rejected") as unknown as K8sObject;
|
|
}
|
|
|
|
async function kubernetesGetPodEvents(options: RunnerRetentionOptions): Promise<K8sObject[]> {
|
|
const result = await retentionKubernetes(options).list("event", { fieldSelector: "involvedObject.kind=Pod,type=Warning" });
|
|
if (result.code !== 0) throw new AgentRunError("infra-failed", "Kubernetes get events failed", { httpStatus: 502, details: retentionFailureDetails(result, { reason: "runner-kubernetes-observation-failed", namespace: options.namespace, resource: "events" }) });
|
|
const parsed = parseJsonObject(result.stdout, "Kubernetes 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) : [];
|
|
}
|
|
|
|
async function deleteAssociatedResources(options: RunnerRetentionOptions, runner: RunnerResourceEntry): Promise<number> {
|
|
let deleted = 0;
|
|
const selector = `agentrun.pikastech.local/runner-job=${runner.jobName}`;
|
|
const secrets = await kubernetesGetAssociatedResources(options, "secret", selector, runner.jobName);
|
|
for (const secret of secrets) {
|
|
const name = objectName(secret);
|
|
if (!name || !isExclusiveRunnerSecret(secret, runner)) {
|
|
throw new AgentRunError("infra-failed", "runner retention associated Secret ownership is not exclusive", {
|
|
httpStatus: 409,
|
|
details: {
|
|
reason: "runner-associated-resource-ownership-unknown",
|
|
namespace: options.namespace,
|
|
resource: "secret",
|
|
name,
|
|
jobName: runner.jobName,
|
|
valuesPrinted: false,
|
|
},
|
|
});
|
|
}
|
|
const result = await retentionKubernetes(options).delete("secret", name, { ignoreNotFound: true, propagationPolicy: "Background" });
|
|
if (result.code !== 0) {
|
|
throw new AgentRunError("infra-failed", "Kubernetes delete associated secret failed", {
|
|
httpStatus: 502,
|
|
details: retentionFailureDetails(result, { reason: "runner-associated-resource-delete-failed", namespace: options.namespace, resource: "secret", name, jobName: runner.jobName }),
|
|
});
|
|
}
|
|
if (result.transport === "in-cluster-https" ? result.kubernetesReason !== "NotFound" : countDeletedLines(result.stdout) > 0) deleted++;
|
|
}
|
|
return deleted;
|
|
}
|
|
|
|
async function kubernetesGetAssociatedResources(options: RunnerRetentionOptions, resource: "secret", selector: string, jobName: string): Promise<K8sObject[]> {
|
|
const result = await retentionKubernetes(options).list(resource, { labelSelector: selector });
|
|
if (result.code !== 0) {
|
|
throw new AgentRunError("infra-failed", `Kubernetes get associated ${resource} failed`, {
|
|
httpStatus: 502,
|
|
details: retentionFailureDetails(result, { reason: "runner-associated-resource-observation-failed", namespace: options.namespace, resource, jobName }),
|
|
});
|
|
}
|
|
const parsed = parseJsonObject(result.stdout, `Kubernetes get associated ${resource}`, "runner-associated-resource-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)
|
|
: [];
|
|
}
|
|
|
|
function isExclusiveRunnerSecret(secret: K8sObject, runner: RunnerResourceEntry): boolean {
|
|
const labels = secret.metadata?.labels ?? {};
|
|
const annotations = secret.metadata?.annotations ?? {};
|
|
const identity = {
|
|
"agentrun.pikastech.local/run-id": runner.runId ?? "",
|
|
"agentrun.pikastech.local/command-id": runner.commandId ?? "",
|
|
"agentrun.pikastech.local/runner-id": runner.runnerId ?? "",
|
|
"agentrun.pikastech.local/runner-job": runner.jobName,
|
|
};
|
|
if (labels["app.kubernetes.io/component"] !== "runner-transient-env") return false;
|
|
if (labels["agentrun.pikastech.local/runner-job"] !== runner.jobName) return false;
|
|
if (annotations["agentrun.pikastech.local/runner-job"] !== runner.jobName) return false;
|
|
if (!resourceIdentityMatches(identity, annotations)) return false;
|
|
const owners = secret.metadata?.ownerReferences ?? [];
|
|
return owners.length === 0 || owners.every((owner) => owner.kind === "Job" && owner.name === runner.jobName);
|
|
}
|
|
|
|
function retentionKubernetes(options: RunnerRetentionOptions): KubernetesClient {
|
|
return createKubernetesClient({ namespace: options.namespace, ...(options.kubectlCommand ? { kubectlCommand: options.kubectlCommand } : {}), ...(options.signal ? { signal: options.signal } : {}) });
|
|
}
|
|
|
|
function retentionResource(resource: string): KubernetesResource {
|
|
if (resource === "jobs" || resource === "job") return "job";
|
|
if (resource === "pods" || resource === "pod") return "pod";
|
|
throw new AgentRunError("schema-invalid", `runner retention resource ${resource} is unsupported`, { httpStatus: 500, details: { reason: "runner-retention-resource-unsupported", resource, valuesPrinted: false } });
|
|
}
|
|
|
|
function retentionFailureDetails(result: KubernetesTransportResult, context: JsonRecord): JsonRecord {
|
|
return redactJson({
|
|
...context,
|
|
exitCode: result.code,
|
|
signal: result.signal,
|
|
transport: result.transport,
|
|
statusCode: result.statusCode,
|
|
kubernetesReason: result.kubernetesReason,
|
|
stderr: redactText(result.stderr.slice(-2000)),
|
|
stdoutHash: result.stdout ? stableHash(result.stdout) : null,
|
|
valuesPrinted: false,
|
|
});
|
|
}
|
|
|
|
function compareCandidates(left: RunnerResourceEntry, right: RunnerResourceEntry): number {
|
|
const priority = (kind: CandidateKind): number => kind === "idle" ? 0 : kind === "terminal" ? 1 : kind === "stale-pending" ? 2 : 3;
|
|
const byKind = priority(left.candidateKind) - priority(right.candidateKind);
|
|
if (byKind !== 0) return byKind;
|
|
const byActive = left.lastActiveAtMs - right.lastActiveAtMs;
|
|
if (byActive !== 0) return byActive;
|
|
return left.name.localeCompare(right.name);
|
|
}
|
|
|
|
function resourceSummary(item: RunnerResourceEntry): JsonRecord {
|
|
return redactJson({
|
|
resourceKind: item.resourceKind,
|
|
name: item.name,
|
|
jobName: item.jobName,
|
|
resourceUid: item.resourceUid,
|
|
resourceVersion: item.resourceVersion,
|
|
runId: item.runId,
|
|
commandId: item.commandId,
|
|
runnerId: item.runnerId,
|
|
protectedReason: item.protectedReason,
|
|
candidateKind: item.candidateKind,
|
|
classificationReason: item.classificationReason,
|
|
repairKind: item.repairKind,
|
|
terminal: item.terminal,
|
|
podPhase: item.podPhase,
|
|
podObservations: item.podObservations.slice(0, 5).map(podObservationSummary),
|
|
podObservationOmittedCount: Math.max(0, item.podObservations.length - 5),
|
|
lastActiveAt: new Date(item.lastActiveAtMs).toISOString(),
|
|
valuesPrinted: false,
|
|
}) as JsonRecord;
|
|
}
|
|
|
|
function podObservationSummary(item: RunnerPodObservation): JsonRecord {
|
|
return {
|
|
...item,
|
|
waiting: item.waiting.slice(0, 5),
|
|
waitingOmittedCount: Math.max(0, item.waiting.length - 5),
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function cleanupPolicySummary(): JsonRecord {
|
|
return {
|
|
mode: "db-ledger-and-k8s-observation",
|
|
forceActive: false,
|
|
activeProtection: "protect fresh claims and active Pods; terminal Pods do not remain active merely because their container once started",
|
|
terminalClaimRepair: "release stale terminal run claims under the same DB and Kubernetes CAS fence",
|
|
deletionOrder: "idle, terminal, stale-pending, inactive by lastActiveAt",
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function labelSelector(labels: Record<string, string>): string {
|
|
return Object.entries(labels).map(([key, value]) => `${key}=${value}`).join(",");
|
|
}
|
|
|
|
function matchesJobPrefix(name: string, prefixes: string[]): boolean {
|
|
return prefixes.length === 0 || prefixes.some((prefix) => name === prefix || name.startsWith(`${prefix}-`));
|
|
}
|
|
|
|
function podJobName(item: K8sObject): string | null {
|
|
const labels = item.metadata?.labels ?? {};
|
|
if (labels["job-name"]) return labels["job-name"];
|
|
return item.metadata?.ownerReferences?.find((owner) => owner.kind === "Job")?.name ?? null;
|
|
}
|
|
|
|
function resourceIdentityMatches(left: Record<string, string>, right: Record<string, string>): boolean {
|
|
for (const key of ["agentrun.pikastech.local/run-id", "agentrun.pikastech.local/command-id", "agentrun.pikastech.local/runner-id"] as const) {
|
|
if (!left[key] || !right[key] || left[key] !== right[key]) return false;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
function jobTerminal(item: K8sObject): boolean {
|
|
const conditions = item.status?.conditions;
|
|
if (!Array.isArray(conditions)) return false;
|
|
return conditions.some((condition) => {
|
|
const record = condition as JsonRecord;
|
|
return (record.type === "Complete" || record.type === "Failed") && record.status === "True";
|
|
});
|
|
}
|
|
|
|
function podTerminal(item: K8sObject): boolean {
|
|
const phase = stringPath(item, ["status", "phase"]);
|
|
return phase === "Succeeded" || phase === "Failed";
|
|
}
|
|
|
|
function observePod(item: K8sObject, stalePendingMaxAgeMs: number, events: K8sObject[], eventFactsAvailable: boolean): RunnerPodObservation {
|
|
const createdAtMs = objectCreatedAtMs(item);
|
|
const ageMs = createdAtMs > 0 ? Math.max(0, Date.now() - createdAtMs) : null;
|
|
const phase = stringPath(item, ["status", "phase"]);
|
|
const conditions = jsonRecordArray(item.status?.conditions);
|
|
const statuses = jsonRecordArray(item.status?.containerStatuses);
|
|
const runner = statuses.find((status) => status.name === "runner") ?? null;
|
|
const runnerState = jsonRecord(runner?.state);
|
|
const runnerContainerState: RunnerPodObservation["runnerContainerState"] = runnerState?.waiting ? "waiting" : runnerState?.running ? "running" : runnerState?.terminated ? "terminated" : "unknown";
|
|
const ready = conditions.some((condition) => condition.type === "Ready" && condition.status === "True");
|
|
const waiting = [
|
|
...statuses.flatMap((status) => waitingSummary(status, "container", typeof status.name === "string" ? status.name : null)),
|
|
...jsonRecordArray(item.status?.initContainerStatuses).flatMap((status) => waitingSummary(status, "init-container", typeof status.name === "string" ? status.name : null)),
|
|
...conditions.flatMap((condition) => condition.status === "False" && typeof condition.reason === "string" && condition.reason.length > 0
|
|
? [{ source: "condition", name: typeof condition.type === "string" ? condition.type : null, reason: condition.reason, message: boundedRedactedText(condition.message), deterministic: condition.reason === "Unschedulable", valuesPrinted: false }]
|
|
: []),
|
|
...events.flatMap(eventWaitingSummary),
|
|
];
|
|
const runnerContainerRestartCount = typeof runner?.restartCount === "number" ? runner.restartCount : 0;
|
|
const runnerLastState = jsonRecord(runner?.lastState);
|
|
const runnerContainerLastState: RunnerPodObservation["runnerContainerLastState"] = runnerLastState?.running ? "running" : runnerLastState?.terminated ? "terminated" : "none";
|
|
const runnerContainerStarted = runner?.started === true || runnerContainerRestartCount > 0 || runnerContainerState === "running" || runnerContainerState === "terminated" || Boolean(runnerLastState?.running) || Boolean(runnerLastState?.terminated);
|
|
const runnerContainerReady = runner?.ready === true;
|
|
return {
|
|
name: objectName(item) ?? "unknown",
|
|
uid: item.metadata?.uid ?? null,
|
|
resourceVersion: item.metadata?.resourceVersion ?? null,
|
|
phase,
|
|
createdAt: createdAtMs > 0 ? new Date(createdAtMs).toISOString() : null,
|
|
ageMs,
|
|
stale: ageMs !== null && ageMs >= stalePendingMaxAgeMs,
|
|
podStartTime: boundedRedactedText(item.status?.startTime),
|
|
ready,
|
|
runnerContainerState,
|
|
runnerContainerLastState,
|
|
runnerContainerStarted,
|
|
runnerContainerReady,
|
|
runnerContainerRestartCount,
|
|
waiting,
|
|
waitingEvidenceSources: [...new Set(waiting.map((entry) => typeof entry.source === "string" ? entry.source : "unknown"))],
|
|
eventFactsAvailable,
|
|
factsAvailable: phase !== null && createdAtMs > 0,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function waitingSummary(status: JsonRecord, source: string, name: string | null): JsonRecord[] {
|
|
const state = jsonRecord(status.state);
|
|
const waiting = jsonRecord(state?.waiting);
|
|
if (!waiting || typeof waiting.reason !== "string" || waiting.reason.length === 0) return [];
|
|
return [{ source, name, reason: waiting.reason, message: boundedRedactedText(waiting.message), deterministic: waiting.reason !== "ContainerCreating" && waiting.reason !== "PodInitializing", valuesPrinted: false }];
|
|
}
|
|
|
|
function eventWaitingSummary(event: K8sObject): JsonRecord[] {
|
|
if (event.type !== "Warning" || typeof event.reason !== "string" || event.reason.length === 0) return [];
|
|
return [{
|
|
source: "event",
|
|
name: typeof event.metadata?.name === "string" ? event.metadata.name : null,
|
|
reason: event.reason,
|
|
message: boundedRedactedText(event.message),
|
|
deterministic: true,
|
|
count: typeof event.count === "number" ? event.count : null,
|
|
lastObservedAt: boundedRedactedText(event.series?.lastObservedTime ?? event.eventTime ?? event.lastTimestamp ?? event.firstTimestamp),
|
|
valuesPrinted: false,
|
|
}];
|
|
}
|
|
|
|
function podEventName(item: K8sObject): string | null {
|
|
if (item.involvedObject?.kind !== "Pod") return null;
|
|
return typeof item.involvedObject.name === "string" && item.involvedObject.name.length > 0 ? item.involvedObject.name : null;
|
|
}
|
|
|
|
function stalePendingPodBlockReason(observations: RunnerPodObservation[]): string | null {
|
|
if (observations.length === 0) return "pod-facts-unavailable";
|
|
if (observations.some((item) => !item.factsAvailable)) return "pod-facts-unavailable";
|
|
const activeReason = activePodProtectionReason(observations);
|
|
if (activeReason) return activeReason;
|
|
if (observations.some((item) => item.phase !== "Pending")) return "runner-pod-not-pending";
|
|
if (observations.some((item) => !item.stale)) return "stale-pending-window-active";
|
|
if (observations.some((item) => !item.waiting.some((entry) => entry.deterministic === true) && !item.eventFactsAvailable)) return "pod-waiting-event-facts-unavailable";
|
|
if (observations.some((item) => item.waiting.length > 0 && !item.waiting.some((entry) => entry.deterministic === true))) return "pod-deterministic-waiting-reason-unavailable";
|
|
if (observations.some((item) => item.waiting.length === 0)) return "pod-waiting-reason-unavailable";
|
|
return null;
|
|
}
|
|
|
|
function activePodProtectionReason(observations: RunnerPodObservation[]): string | null {
|
|
if (observations.some((item) => {
|
|
const terminalPhase = item.phase === "Succeeded" || item.phase === "Failed";
|
|
return item.phase === "Running"
|
|
|| item.ready
|
|
|| item.runnerContainerReady
|
|
|| item.runnerContainerState === "running"
|
|
|| (!terminalPhase && (item.runnerContainerStarted || item.runnerContainerState === "terminated"));
|
|
})) return "runner-pod-started-running-or-ready";
|
|
return null;
|
|
}
|
|
|
|
function hasNoBusinessProgress(run: RunRecord, command: CommandRecord, commands: CommandRecord[], events: import("../common/types.js").RunEvent[], stalePendingMaxAgeMs: number): boolean {
|
|
const cutoff = Date.now() - stalePendingMaxAgeMs;
|
|
if (run.status !== "pending" || run.claimedBy !== null || run.leaseExpiresAt !== null || run.cancelRequestId !== null || run.cancelRequestedAt !== null) return false;
|
|
if (command.state !== "pending" || command.acknowledgedAt !== null || command.cancelRequestId !== null || command.cancelRequestedAt !== null) return false;
|
|
if (!sameValidTimestamp(run.createdAt, run.updatedAt) || !sameValidTimestamp(command.createdAt, command.updatedAt)) return false;
|
|
if (isoMs(run.createdAt) > cutoff || isoMs(command.createdAt) > cutoff) return false;
|
|
if (commands.length !== 1 || commands[0]?.id !== command.id) return false;
|
|
if (events.length === 0 || events.length >= 500) return false;
|
|
const allowedPhases = new Set(["run-created", "command-created", "runner-job-created", "runner-dispatch-completed"]);
|
|
const userMessages = events.filter((event) => event.type === "user_message");
|
|
if (userMessages.length > 1) return false;
|
|
const phases = events.flatMap((event) => {
|
|
if (event.type === "user_message") {
|
|
const payload = event.payload;
|
|
const valid = payload.commandId === command.id
|
|
&& typeof payload.userMessageId === "string" && payload.userMessageId.length > 0
|
|
&& typeof payload.text === "string" && payload.text.length > 0
|
|
&& typeof payload.source === "string" && payload.source.length > 0;
|
|
return valid ? [] : [null];
|
|
}
|
|
const phase = typeof event.payload.phase === "string" ? event.payload.phase : null;
|
|
return event.type === "backend_status" && phase !== null && allowedPhases.has(phase) ? [phase] : [null];
|
|
});
|
|
if (phases.includes(null)) return false;
|
|
if (!phases.includes("run-created") || !phases.includes("command-created") || !phases.includes("runner-job-created")) return false;
|
|
return events.every((event) => event.payload.commandId === undefined || event.payload.commandId === command.id);
|
|
}
|
|
|
|
function protectedActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string | null): ActivityFields {
|
|
return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: reason, candidateKind: "inactive", classificationReason: reason, repairKind: "none", lastActiveAtMs, dbFactsFingerprint };
|
|
}
|
|
|
|
function candidateActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, kind: CandidateKind, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string, repairKind: RepairKind = "none"): ActivityFields {
|
|
return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind: kind, classificationReason: reason, repairKind, lastActiveAtMs, dbFactsFingerprint };
|
|
}
|
|
|
|
function databaseFactsFingerprint(run: RunRecord, command: CommandRecord, commands: CommandRecord[], events: import("../common/types.js").RunEvent[]): string {
|
|
const commandFacts = (item: CommandRecord): JsonRecord => ({
|
|
id: item.id,
|
|
runId: item.runId,
|
|
seq: item.seq,
|
|
type: item.type,
|
|
state: item.state,
|
|
payloadHash: item.payloadHash,
|
|
cancelEpoch: item.cancelEpoch,
|
|
cancelRequestId: item.cancelRequestId,
|
|
cancelRequestedAt: item.cancelRequestedAt,
|
|
cancelReason: item.cancelReason,
|
|
createdAt: item.createdAt,
|
|
updatedAt: item.updatedAt,
|
|
acknowledgedAt: item.acknowledgedAt,
|
|
dispatchIntentHash: stableHash(item.dispatchIntent ?? null),
|
|
valuesPrinted: false,
|
|
});
|
|
return stableHash({
|
|
run: {
|
|
id: run.id,
|
|
status: run.status,
|
|
terminalStatus: run.terminalStatus,
|
|
failureKind: run.failureKind,
|
|
failureMessageHash: run.failureMessage ? stableHash(run.failureMessage) : null,
|
|
cancelEpoch: run.cancelEpoch,
|
|
cancelRequestId: run.cancelRequestId,
|
|
cancelRequestedAt: run.cancelRequestedAt,
|
|
cancelReason: run.cancelReason,
|
|
createdAt: run.createdAt,
|
|
updatedAt: run.updatedAt,
|
|
claimedBy: run.claimedBy,
|
|
leaseExpiresAt: run.leaseExpiresAt,
|
|
valuesPrinted: false,
|
|
},
|
|
command: commandFacts(command),
|
|
commands: [...commands].sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)).map(commandFacts),
|
|
events: [...events].sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)).map((event) => ({
|
|
id: event.id,
|
|
runId: event.runId,
|
|
seq: event.seq,
|
|
type: event.type,
|
|
createdAt: event.createdAt,
|
|
payloadHash: stableHash(event.payload),
|
|
valuesPrinted: false,
|
|
})),
|
|
valuesPrinted: false,
|
|
});
|
|
}
|
|
|
|
function runnerSafetyFingerprint(item: RunnerResourceEntry): string {
|
|
return stableHash({
|
|
schema: "agentrun-runner-retention-safety-v1",
|
|
resourceKind: item.resourceKind,
|
|
name: item.name,
|
|
jobName: item.jobName,
|
|
resourceUid: item.resourceUid,
|
|
resourceVersion: item.resourceVersion,
|
|
runId: item.runId,
|
|
commandId: item.commandId,
|
|
runnerId: item.runnerId,
|
|
createdAtMs: item.createdAtMs,
|
|
terminal: item.terminal,
|
|
podPhase: item.podPhase,
|
|
protectedActive: item.protectedActive,
|
|
protectedReason: item.protectedReason,
|
|
candidateKind: item.candidateKind,
|
|
classificationReason: item.classificationReason,
|
|
repairKind: item.repairKind,
|
|
lastActiveAtMs: item.lastActiveAtMs,
|
|
dbFactsFingerprint: item.dbFactsFingerprint,
|
|
podObservations: [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name)).map(podSafetyFacts),
|
|
valuesPrinted: false,
|
|
});
|
|
}
|
|
|
|
function podSafetyFacts(item: RunnerPodObservation): JsonRecord {
|
|
const waiting = item.waiting.map((entry) => ({
|
|
source: typeof entry.source === "string" ? entry.source : null,
|
|
name: typeof entry.name === "string" ? entry.name : null,
|
|
reason: typeof entry.reason === "string" ? entry.reason : null,
|
|
deterministic: entry.deterministic === true,
|
|
valuesPrinted: false,
|
|
})).sort((left, right) => JSON.stringify(left).localeCompare(JSON.stringify(right)));
|
|
return {
|
|
name: item.name,
|
|
uid: item.uid,
|
|
resourceVersion: item.resourceVersion,
|
|
phase: item.phase,
|
|
createdAt: item.createdAt,
|
|
stale: item.stale,
|
|
podStartTime: item.podStartTime,
|
|
ready: item.ready,
|
|
runnerContainerState: item.runnerContainerState,
|
|
runnerContainerLastState: item.runnerContainerLastState,
|
|
runnerContainerStarted: item.runnerContainerStarted,
|
|
runnerContainerReady: item.runnerContainerReady,
|
|
runnerContainerRestartCount: item.runnerContainerRestartCount,
|
|
waiting,
|
|
waitingEvidenceSources: [...item.waitingEvidenceSources].sort(),
|
|
eventFactsAvailable: item.eventFactsAvailable,
|
|
factsAvailable: item.factsAvailable,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function boundedResourceSummaries(items: RunnerResourceEntry[]): { items: JsonRecord[]; omittedCount: number } {
|
|
const limit = 20;
|
|
return { items: items.slice(0, limit).map(resourceSummary), omittedCount: Math.max(0, items.length - limit) };
|
|
}
|
|
|
|
function boundedJsonRecords(items: JsonRecord[]): { items: JsonRecord[]; omittedCount: number } {
|
|
const limit = 20;
|
|
return { items: items.slice(0, limit), omittedCount: Math.max(0, items.length - limit) };
|
|
}
|
|
|
|
function resourceIdentitySummary(item: RunnerResourceEntry): JsonRecord {
|
|
return {
|
|
resourceKind: item.resourceKind,
|
|
name: item.name,
|
|
jobName: item.jobName,
|
|
runId: item.runId,
|
|
commandId: item.commandId,
|
|
runnerId: item.runnerId,
|
|
candidateKind: item.candidateKind,
|
|
classificationReason: item.classificationReason,
|
|
repairKind: item.repairKind,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function retentionErrorSummary(error: unknown): JsonRecord {
|
|
if (error instanceof AgentRunError) {
|
|
return redactJson({
|
|
failureKind: error.failureKind,
|
|
httpStatus: error.httpStatus,
|
|
message: redactText(error.message).slice(0, 300),
|
|
details: error.details,
|
|
valuesPrinted: false,
|
|
}) as JsonRecord;
|
|
}
|
|
return {
|
|
failureKind: "infra-failed",
|
|
httpStatus: 500,
|
|
message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 300),
|
|
details: null,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function reasonCounts(reasons: string[]): JsonRecord {
|
|
const counts: Record<string, number> = {};
|
|
for (const reason of reasons) counts[reason] = (counts[reason] ?? 0) + 1;
|
|
return Object.fromEntries(Object.entries(counts).sort(([left], [right]) => left.localeCompare(right))) as JsonRecord;
|
|
}
|
|
|
|
function clampReconcileLimit(limit: number): number {
|
|
return Math.max(1, Math.min(Math.floor(limit), 500));
|
|
}
|
|
|
|
function jsonRecord(value: unknown): JsonRecord | null {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : null;
|
|
}
|
|
|
|
function jsonRecordArray(value: unknown): JsonRecord[] {
|
|
return Array.isArray(value) ? value.map(jsonRecord).filter((item): item is JsonRecord => item !== null) : [];
|
|
}
|
|
|
|
function boundedRedactedText(value: unknown): string | null {
|
|
if (typeof value !== "string" || value.length === 0) return null;
|
|
return redactText(value).slice(0, 512);
|
|
}
|
|
|
|
function sameValidTimestamp(left: string, right: string): boolean {
|
|
return isoMs(left) > 0 && isoMs(left) === isoMs(right);
|
|
}
|
|
|
|
function heartbeatFresh(leaseExpiresAt: string | null, activeHeartbeatMaxAgeMs: number): boolean {
|
|
const leaseMs = isoMs(leaseExpiresAt);
|
|
if (leaseMs <= 0) return false;
|
|
return leaseMs + activeHeartbeatMaxAgeMs >= Date.now();
|
|
}
|
|
|
|
function objectName(item: K8sObject): string | null {
|
|
return typeof item.metadata?.name === "string" && item.metadata.name.length > 0 ? item.metadata.name : null;
|
|
}
|
|
|
|
function objectCreatedAtMs(item: K8sObject): number {
|
|
return isoMs(item.metadata?.creationTimestamp) || 0;
|
|
}
|
|
|
|
function isoMs(value: string | null | undefined): number {
|
|
if (!value) return 0;
|
|
const parsed = Date.parse(value);
|
|
return Number.isFinite(parsed) ? parsed : 0;
|
|
}
|
|
|
|
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 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: { reason, 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 {
|
|
return stdout.split(/\r?\n/u).filter((line) => /\sdeleted$/u.test(line.trim())).length;
|
|
}
|