fix: 收紧 durable dispatch 执行边界
This commit is contained in:
root
@@ -0,0 +1,49 @@
|
||||
import { AgentRunError } from "./errors.js";
|
||||
import type { AipodImageRef, JsonRecord } from "./types.js";
|
||||
|
||||
export function validateAipodImageRef(value: unknown, fieldName = "imageRef"): AipodImageRef {
|
||||
const record = asRecord(value, fieldName);
|
||||
const kind = requiredString(record, "kind", fieldName);
|
||||
if (kind !== "env-image-dockerfile") throw new AgentRunError("schema-invalid", `${fieldName}.kind must be env-image-dockerfile`, { httpStatus: 400 });
|
||||
const repoUrl = validateRepoUrl(requiredString(record, "repoUrl", fieldName), `${fieldName}.repoUrl`);
|
||||
const commitId = requiredString(record, "commitId", fieldName).toLowerCase();
|
||||
if (!/^[0-9a-f]{40}$/u.test(commitId)) throw new AgentRunError("schema-invalid", `${fieldName}.commitId must be a full 40-character git commit sha`, { httpStatus: 400 });
|
||||
const dockerfilePath = validateDockerfilePath(requiredString(record, "dockerfilePath", fieldName), `${fieldName}.dockerfilePath`);
|
||||
return { kind: "env-image-dockerfile", repoUrl, commitId, dockerfilePath };
|
||||
}
|
||||
|
||||
function validateRepoUrl(value: string, fieldName: string): string {
|
||||
if (/\s|[\x00-\x1f\x7f]/u.test(value)) throw new AgentRunError("schema-invalid", `${fieldName} must not contain whitespace or control characters`, { httpStatus: 400 });
|
||||
if (value.includes("://")) {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(value);
|
||||
} catch {
|
||||
throw new AgentRunError("schema-invalid", `${fieldName} must be a valid git repo URL`, { httpStatus: 400 });
|
||||
}
|
||||
if (!["https:", "http:", "ssh:", "git:"].includes(url.protocol)) throw new AgentRunError("schema-invalid", `${fieldName} must use http(s), ssh, or git protocol`, { httpStatus: 400 });
|
||||
if (url.password || (url.username && !(url.protocol === "ssh:" && url.username === "git"))) throw new AgentRunError("schema-invalid", `${fieldName} must not include credentials`, { httpStatus: 400 });
|
||||
if (url.search || url.hash) throw new AgentRunError("schema-invalid", `${fieldName} must not include query or fragment`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
if (!/^[A-Za-z0-9._-]+@[A-Za-z0-9._-]+:[A-Za-z0-9._/-]+(?:\.git)?$/u.test(value)) throw new AgentRunError("schema-invalid", `${fieldName} must be a git repo URL`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
|
||||
function validateDockerfilePath(value: string, fieldName: string): string {
|
||||
if (value === "." || value.startsWith("/") || value.endsWith("/") || value.includes("\\")) throw new AgentRunError("schema-invalid", `${fieldName} must be a repository-relative file path`, { httpStatus: 400 });
|
||||
const parts = value.split("/");
|
||||
if (parts.some((part) => part.length === 0 || part === "." || part === "..")) throw new AgentRunError("schema-invalid", `${fieldName} must stay within the checkout`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
|
||||
function requiredString(record: JsonRecord, key: string, fieldName: string): string {
|
||||
const value = record[key];
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${fieldName}.${key} is required`, { httpStatus: 400 });
|
||||
return value.trim();
|
||||
}
|
||||
|
||||
function asRecord(value: unknown, fieldName: string): JsonRecord {
|
||||
if (typeof value === "object" && value !== null && !Array.isArray(value)) return value as JsonRecord;
|
||||
throw new AgentRunError("schema-invalid", `${fieldName} must be an object`, { httpStatus: 400 });
|
||||
}
|
||||
@@ -1,7 +1,10 @@
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { AgentRunError } from "./errors.js";
|
||||
import type { AipodImageRef, JsonRecord } from "./types.js";
|
||||
import { asRecord, stableHash } from "./validation.js";
|
||||
import { stableHash } from "./validation.js";
|
||||
import { validateAipodImageRef } from "./aipod-image-ref-validation.js";
|
||||
|
||||
export { validateAipodImageRef } from "./aipod-image-ref-validation.js";
|
||||
|
||||
export interface RunnerEnvImageResolution extends JsonRecord {
|
||||
status: "explicit-image" | "catalog-reused" | "runtime-default-reused" | "legacy-default";
|
||||
@@ -13,17 +16,6 @@ export interface RunnerEnvImageResolution extends JsonRecord {
|
||||
valuesPrinted: false;
|
||||
}
|
||||
|
||||
export function validateAipodImageRef(value: unknown, fieldName = "imageRef"): AipodImageRef {
|
||||
const record = asRecord(value, fieldName);
|
||||
const kind = requiredString(record, "kind", fieldName);
|
||||
if (kind !== "env-image-dockerfile") throw new AgentRunError("schema-invalid", `${fieldName}.kind must be env-image-dockerfile`, { httpStatus: 400 });
|
||||
const repoUrl = validateRepoUrl(requiredString(record, "repoUrl", fieldName), `${fieldName}.repoUrl`);
|
||||
const commitId = requiredString(record, "commitId", fieldName).toLowerCase();
|
||||
if (!/^[0-9a-f]{40}$/u.test(commitId)) throw new AgentRunError("schema-invalid", `${fieldName}.commitId must be a full 40-character git commit sha`, { httpStatus: 400 });
|
||||
const dockerfilePath = validateDockerfilePath(requiredString(record, "dockerfilePath", fieldName), `${fieldName}.dockerfilePath`);
|
||||
return { kind: "env-image-dockerfile", repoUrl, commitId, dockerfilePath };
|
||||
}
|
||||
|
||||
export function imageRefSourceSummary(imageRef: AipodImageRef): JsonRecord {
|
||||
return {
|
||||
kind: imageRef.kind,
|
||||
@@ -136,37 +128,6 @@ function sameImageRef(left: AipodImageRef, right: AipodImageRef): boolean {
|
||||
return left.kind === right.kind && left.repoUrl === right.repoUrl && left.commitId === right.commitId && left.dockerfilePath === right.dockerfilePath;
|
||||
}
|
||||
|
||||
function validateRepoUrl(value: string, fieldName: string): string {
|
||||
if (/\s|[\x00-\x1f\x7f]/u.test(value)) throw new AgentRunError("schema-invalid", `${fieldName} must not contain whitespace or control characters`, { httpStatus: 400 });
|
||||
if (value.includes("://")) {
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(value);
|
||||
} catch {
|
||||
throw new AgentRunError("schema-invalid", `${fieldName} must be a valid git repo URL`, { httpStatus: 400 });
|
||||
}
|
||||
if (!["https:", "http:", "ssh:", "git:"].includes(url.protocol)) throw new AgentRunError("schema-invalid", `${fieldName} must use http(s), ssh, or git protocol`, { httpStatus: 400 });
|
||||
if (url.password || (url.username && !(url.protocol === "ssh:" && url.username === "git"))) throw new AgentRunError("schema-invalid", `${fieldName} must not include credentials`, { httpStatus: 400 });
|
||||
if (url.search || url.hash) throw new AgentRunError("schema-invalid", `${fieldName} must not include query or fragment`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
if (!/^[A-Za-z0-9._-]+@[A-Za-z0-9._-]+:[A-Za-z0-9._/-]+(?:\.git)?$/u.test(value)) throw new AgentRunError("schema-invalid", `${fieldName} must be a git repo URL`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
|
||||
function validateDockerfilePath(value: string, fieldName: string): string {
|
||||
if (value === "." || value.startsWith("/") || value.endsWith("/") || value.includes("\\")) throw new AgentRunError("schema-invalid", `${fieldName} must be a repository-relative file path`, { httpStatus: 400 });
|
||||
const parts = value.split("/");
|
||||
if (parts.some((part) => part.length === 0 || part === "." || part === "..")) throw new AgentRunError("schema-invalid", `${fieldName} must stay within the checkout`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
|
||||
function requiredString(record: JsonRecord, key: string, fieldName: string): string {
|
||||
const value = record[key];
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${fieldName}.${key} is required`, { httpStatus: 400 });
|
||||
return value.trim();
|
||||
}
|
||||
|
||||
function stringValue(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
||||
}
|
||||
|
||||
@@ -240,6 +240,13 @@ export interface RunnerDispatchRequest extends JsonRecord {
|
||||
}
|
||||
|
||||
export type RunnerDispatchIntentState = "pending" | "dispatching" | "retry" | "dispatched" | "failed" | "cancelled";
|
||||
export type RunnerDispatchOutcome = "kubernetes-job" | "active-runner-reused";
|
||||
|
||||
export interface RunnerDispatchCompletion {
|
||||
dispatchOutcome: RunnerDispatchOutcome;
|
||||
actualRunnerJobId: string | null;
|
||||
activeRunnerId: string | null;
|
||||
}
|
||||
|
||||
export interface RunnerDispatchIntentRecord extends JsonRecord {
|
||||
id: string;
|
||||
@@ -251,6 +258,9 @@ export interface RunnerDispatchIntentRecord extends JsonRecord {
|
||||
state: RunnerDispatchIntentState;
|
||||
attemptCount: number;
|
||||
runnerJobId: string;
|
||||
dispatchOutcome: RunnerDispatchOutcome | null;
|
||||
actualRunnerJobId: string | null;
|
||||
activeRunnerId: string | null;
|
||||
leaseOwner: string | null;
|
||||
leaseExpiresAt: string | null;
|
||||
nextAttemptAt: string;
|
||||
@@ -265,6 +275,9 @@ export interface RunnerDispatchIntentSummary extends JsonRecord {
|
||||
id: string;
|
||||
state: RunnerDispatchIntentState;
|
||||
runnerJobId: string;
|
||||
dispatchOutcome: RunnerDispatchOutcome | null;
|
||||
actualRunnerJobId: string | null;
|
||||
activeRunnerId: string | null;
|
||||
attemptCount: number;
|
||||
durable: true;
|
||||
}
|
||||
|
||||
+132
-13
@@ -2,6 +2,7 @@ import { createHash, randomUUID } from "node:crypto";
|
||||
import type { BackendProfile, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, ExecutionPolicy, JsonRecord, JsonValue, QueueTaskState, ResourceBundleRef, SecretRef, SessionListState, SessionRef } from "./types.js";
|
||||
import { AgentRunError } from "./errors.js";
|
||||
import { backendProfileIdPattern, backendProfileSpec, isBackendProfile } from "./backend-profiles.js";
|
||||
import { validateAipodImageRef } from "./aipod-image-ref-validation.js";
|
||||
|
||||
const backendProfilePatternText = String(backendProfileIdPattern);
|
||||
|
||||
@@ -324,25 +325,143 @@ export function validateCreateCommand(input: unknown): CreateCommandInput {
|
||||
if (type !== "turn") throw new AgentRunError("schema-invalid", "dispatch is only supported for turn commands", { httpStatus: 400 });
|
||||
const dispatchRecord = asRecord(record.dispatch, "command.dispatch");
|
||||
if (requiredString(dispatchRecord, "kind") !== "kubernetes-job") throw new AgentRunError("schema-invalid", "command.dispatch.kind must be kubernetes-job", { httpStatus: 400 });
|
||||
const dispatchInput = asRecord(dispatchRecord.input ?? {}, "command.dispatch.input");
|
||||
assertDurableDispatchContainsNoPlaintextSecret(dispatchInput);
|
||||
const dispatchInput = validateDurableDispatchInput(asRecord(dispatchRecord.input ?? {}, "command.dispatch.input"));
|
||||
dispatch = { kind: "kubernetes-job", input: dispatchInput };
|
||||
}
|
||||
return { type, payload, ...(idempotencyKey ? { idempotencyKey } : {}), ...(dispatch ? { dispatch } : {}) };
|
||||
}
|
||||
|
||||
function assertDurableDispatchContainsNoPlaintextSecret(input: JsonRecord): void {
|
||||
if (input.transientEnv === undefined) return;
|
||||
if (!Array.isArray(input.transientEnv)) throw new AgentRunError("schema-invalid", "command.dispatch.input.transientEnv must be an array", { httpStatus: 400 });
|
||||
for (const [index, raw] of input.transientEnv.entries()) {
|
||||
const entry = asRecord(raw, `command.dispatch.input.transientEnv[${index}]`);
|
||||
if (entry.sensitive === true) {
|
||||
throw new AgentRunError("tenant-policy-denied", "durable dispatch cannot persist sensitive transientEnv values; use executionPolicy.toolCredentials SecretRef", {
|
||||
httpStatus: 400,
|
||||
details: { index, name: typeof entry.name === "string" ? entry.name : null, valuesPrinted: false },
|
||||
});
|
||||
}
|
||||
function validateDurableDispatchInput(input: JsonRecord): JsonRecord {
|
||||
assertNoSecretNamedFields(input, "command.dispatch.input");
|
||||
const allowedKeys = new Set([
|
||||
"managerUrl",
|
||||
"image",
|
||||
"namespace",
|
||||
"attemptId",
|
||||
"runnerId",
|
||||
"sourceCommit",
|
||||
"serviceAccountName",
|
||||
"runnerIdleTimeoutMs",
|
||||
"backendRetryMaxAttempts",
|
||||
"backendRetryInitialBackoffMs",
|
||||
"backendRetryMaxBackoffMs",
|
||||
"idempotencyKey",
|
||||
"imageRef",
|
||||
"transientEnv",
|
||||
]);
|
||||
const rejectedKeys = Object.keys(input).filter((key) => !allowedKeys.has(key));
|
||||
if (rejectedKeys.length > 0) throw new AgentRunError("schema-invalid", "command.dispatch.input contains unsupported fields", { httpStatus: 400, details: { rejectedKeys: rejectedKeys.sort(), valuesPrinted: false } });
|
||||
const normalized: JsonRecord = {};
|
||||
for (const key of ["attemptId", "runnerId", "sourceCommit", "idempotencyKey"] as const) {
|
||||
if (input[key] !== undefined) normalized[key] = boundedDispatchString(input[key], `command.dispatch.input.${key}`, 256);
|
||||
}
|
||||
if (input.namespace !== undefined) normalized.namespace = kubernetesName(input.namespace, "command.dispatch.input.namespace");
|
||||
if (input.serviceAccountName !== undefined) normalized.serviceAccountName = kubernetesName(input.serviceAccountName, "command.dispatch.input.serviceAccountName");
|
||||
if (input.managerUrl !== undefined) normalized.managerUrl = credentialFreeManagerUrl(input.managerUrl);
|
||||
if (input.image !== undefined) normalized.image = credentialFreeImage(input.image);
|
||||
for (const key of ["runnerIdleTimeoutMs", "backendRetryMaxAttempts", "backendRetryInitialBackoffMs", "backendRetryMaxBackoffMs"] as const) {
|
||||
if (input[key] !== undefined) normalized[key] = positiveDispatchInteger(input[key], `command.dispatch.input.${key}`);
|
||||
}
|
||||
if (input.imageRef !== undefined) {
|
||||
const imageRefRecord = asRecord(input.imageRef, "command.dispatch.input.imageRef");
|
||||
const extraImageRefKeys = Object.keys(imageRefRecord).filter((key) => !["kind", "repoUrl", "commitId", "dockerfilePath"].includes(key));
|
||||
if (extraImageRefKeys.length > 0) throw new AgentRunError("schema-invalid", "command.dispatch.input.imageRef contains unsupported fields", { httpStatus: 400, details: { rejectedKeys: extraImageRefKeys.sort(), valuesPrinted: false } });
|
||||
normalized.imageRef = validateAipodImageRef(imageRefRecord, "command.dispatch.input.imageRef");
|
||||
}
|
||||
if (normalized.image !== undefined && normalized.imageRef !== undefined) throw new AgentRunError("schema-invalid", "command.dispatch.input.image and imageRef are mutually exclusive", { httpStatus: 400 });
|
||||
if (input.transientEnv !== undefined) normalized.transientEnv = validateDurableTransientEnv(input.transientEnv);
|
||||
return normalized;
|
||||
}
|
||||
|
||||
function validateDurableTransientEnv(value: unknown): JsonRecord[] {
|
||||
if (!Array.isArray(value)) throw new AgentRunError("schema-invalid", "command.dispatch.input.transientEnv must be an array", { httpStatus: 400 });
|
||||
if (value.length > 64) throw new AgentRunError("schema-invalid", "command.dispatch.input.transientEnv must contain at most 64 entries", { httpStatus: 400 });
|
||||
const seen = new Set<string>();
|
||||
return value.map((raw, index) => {
|
||||
const entry = asRecord(raw, `command.dispatch.input.transientEnv[${index}]`);
|
||||
const rejectedEntryKeys = Object.keys(entry).filter((key) => !["name", "value", "sensitive"].includes(key));
|
||||
if (rejectedEntryKeys.length > 0) throw new AgentRunError("schema-invalid", `command.dispatch.input.transientEnv[${index}] contains unsupported fields`, { httpStatus: 400, details: { rejectedKeys: rejectedEntryKeys.sort(), valuesPrinted: false } });
|
||||
const name = requiredString(entry, "name");
|
||||
validateEnvName(name, `command.dispatch.input.transientEnv[${index}].name`);
|
||||
if (isCredentialEnvName(name)) throw durableDispatchSecretError(index, name);
|
||||
if (seen.has(name)) throw new AgentRunError("schema-invalid", `command.dispatch.input.transientEnv name ${name} is duplicated`, { httpStatus: 400 });
|
||||
seen.add(name);
|
||||
const envValue = credentialFreeConfigValue(boundedDispatchString(entry.value, `command.dispatch.input.transientEnv[${index}].value`, 8192, true), `command.dispatch.input.transientEnv[${index}].value`);
|
||||
if (entry.sensitive !== false) throw durableDispatchSecretError(index, name);
|
||||
return { name, value: envValue, sensitive: false };
|
||||
});
|
||||
}
|
||||
|
||||
function assertNoSecretNamedFields(value: unknown, path: string): void {
|
||||
if (Array.isArray(value)) {
|
||||
value.forEach((entry, index) => assertNoSecretNamedFields(entry, `${path}[${index}]`));
|
||||
return;
|
||||
}
|
||||
if (typeof value !== "object" || value === null) return;
|
||||
for (const [key, entry] of Object.entries(value as JsonRecord)) {
|
||||
if (isSecretFieldName(key)) {
|
||||
throw new AgentRunError("tenant-policy-denied", "durable dispatch cannot persist credential-shaped fields; use executionPolicy SecretRef", { httpStatus: 400, details: { field: `${path}.${key}`, valuesPrinted: false } });
|
||||
}
|
||||
assertNoSecretNamedFields(entry, `${path}.${key}`);
|
||||
}
|
||||
}
|
||||
|
||||
function isSecretFieldName(key: string): boolean {
|
||||
const normalized = key.replace(/[^a-z0-9]/giu, "").toLowerCase();
|
||||
return ["secret", "token", "password", "passwd", "apikey", "authorization", "credential", "privatekey", "accesskey", "clientsecret"].some((marker) => normalized.includes(marker)) || normalized === "auth" || normalized.endsWith("auth");
|
||||
}
|
||||
|
||||
function isCredentialEnvName(name: string): boolean {
|
||||
const normalized = name.replace(/[^A-Z0-9]/gu, "");
|
||||
return ["SECRET", "TOKEN", "PASSWORD", "PASSWD", "APIKEY", "AUTH", "CREDENTIAL", "PRIVATEKEY", "ACCESSKEY", "CLIENTSECRET"].some((marker) => normalized.includes(marker));
|
||||
}
|
||||
|
||||
function durableDispatchSecretError(index: number, name: string): AgentRunError {
|
||||
return new AgentRunError("tenant-policy-denied", "durable dispatch cannot persist sensitive transientEnv values; use executionPolicy.toolCredentials SecretRef", { httpStatus: 400, details: { index, name, valuesPrinted: false } });
|
||||
}
|
||||
|
||||
function boundedDispatchString(value: unknown, fieldName: string, maxBytes: number, allowWhitespace = false): string {
|
||||
if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${fieldName} must be a non-empty string`, { httpStatus: 400 });
|
||||
const text = value.trim();
|
||||
if (Buffer.byteLength(text, "utf8") > maxBytes) throw new AgentRunError("schema-invalid", `${fieldName} is too large`, { httpStatus: 400 });
|
||||
if (/[\x00-\x1f\x7f]/u.test(text) || (!allowWhitespace && /\s/u.test(text))) throw new AgentRunError("schema-invalid", `${fieldName} contains unsupported whitespace or control characters`, { httpStatus: 400 });
|
||||
return text;
|
||||
}
|
||||
|
||||
function positiveDispatchInteger(value: unknown, fieldName: string): number {
|
||||
if (!Number.isInteger(value) || Number(value) <= 0) throw new AgentRunError("schema-invalid", `${fieldName} must be a positive integer`, { httpStatus: 400 });
|
||||
return Number(value);
|
||||
}
|
||||
|
||||
function kubernetesName(value: unknown, fieldName: string): string {
|
||||
const name = boundedDispatchString(value, fieldName, 63);
|
||||
if (!/^[a-z0-9](?:[-a-z0-9]*[a-z0-9])?$/u.test(name)) throw new AgentRunError("schema-invalid", `${fieldName} must be a Kubernetes DNS label`, { httpStatus: 400 });
|
||||
return name;
|
||||
}
|
||||
|
||||
function credentialFreeManagerUrl(value: unknown): string {
|
||||
const text = boundedDispatchString(value, "command.dispatch.input.managerUrl", 2048);
|
||||
let url: URL;
|
||||
try { url = new URL(text); } catch { throw new AgentRunError("schema-invalid", "command.dispatch.input.managerUrl must be a valid http(s) URL", { httpStatus: 400 }); }
|
||||
if (!["http:", "https:"].includes(url.protocol) || !url.hostname) throw new AgentRunError("schema-invalid", "command.dispatch.input.managerUrl must be a valid http(s) URL", { httpStatus: 400 });
|
||||
if (url.username || url.password) throw new AgentRunError("tenant-policy-denied", "command.dispatch.input.managerUrl must not include credentials", { httpStatus: 400 });
|
||||
if (url.search || url.hash) throw new AgentRunError("schema-invalid", "command.dispatch.input.managerUrl must not include query or fragment", { httpStatus: 400 });
|
||||
return text;
|
||||
}
|
||||
|
||||
function credentialFreeImage(value: unknown): string {
|
||||
const image = boundedDispatchString(value, "command.dispatch.input.image", 2048);
|
||||
const at = image.lastIndexOf("@");
|
||||
if (at >= 0 && !/^sha256:[0-9a-f]{64}$/u.test(image.slice(at + 1))) throw new AgentRunError("tenant-policy-denied", "command.dispatch.input.image must not include embedded credentials", { httpStatus: 400 });
|
||||
return image;
|
||||
}
|
||||
|
||||
function credentialFreeConfigValue(value: string, fieldName: string): string {
|
||||
if (!value.includes("://")) return value;
|
||||
let url: URL;
|
||||
try { url = new URL(value); } catch { return value; }
|
||||
if (url.username || url.password) throw new AgentRunError("tenant-policy-denied", `${fieldName} must not include URL credentials`, { httpStatus: 400 });
|
||||
return value;
|
||||
}
|
||||
|
||||
function steerPrompt(payload: JsonRecord): string | null {
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
import type { ChildProcess } from "node:child_process";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactText } from "../common/redaction.js";
|
||||
|
||||
export interface ChildProcessExit {
|
||||
code: number | null;
|
||||
signal: NodeJS.Signals | null;
|
||||
}
|
||||
|
||||
export const childProcessKillGraceMs = 250;
|
||||
|
||||
export function throwIfAborted(signal: AbortSignal | undefined, label: string): void {
|
||||
if (signal?.aborted) throw abortedError(signal, label);
|
||||
}
|
||||
|
||||
export function waitForChildProcess(child: ChildProcess, input: { signal?: AbortSignal; label: string; killGraceMs?: number }): Promise<ChildProcessExit> {
|
||||
const killGraceMs = input.killGraceMs ?? childProcessKillGraceMs;
|
||||
return new Promise<ChildProcessExit>((resolve, reject) => {
|
||||
let settled = false;
|
||||
let aborted = false;
|
||||
let killTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
const cleanup = (): void => {
|
||||
if (killTimer) clearTimeout(killTimer);
|
||||
input.signal?.removeEventListener("abort", onAbort);
|
||||
child.removeListener("error", onError);
|
||||
child.removeListener("close", onClose);
|
||||
};
|
||||
const finish = (callback: () => void): void => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
cleanup();
|
||||
callback();
|
||||
};
|
||||
const onAbort = (): void => {
|
||||
if (settled || aborted) return;
|
||||
aborted = true;
|
||||
child.kill("SIGTERM");
|
||||
killTimer = setTimeout(() => {
|
||||
if (!settled) child.kill("SIGKILL");
|
||||
}, killGraceMs);
|
||||
killTimer.unref?.();
|
||||
};
|
||||
const onError = (error: Error): void => {
|
||||
finish(() => reject(aborted ? abortedError(input.signal, input.label) : new AgentRunError("infra-failed", `failed to start ${input.label}: ${redactText(error.message)}`, { httpStatus: 503 })));
|
||||
};
|
||||
const onClose = (code: number | null, signal: NodeJS.Signals | null): void => {
|
||||
finish(() => aborted ? reject(abortedError(input.signal, input.label)) : resolve({ code, signal }));
|
||||
};
|
||||
|
||||
child.once("error", onError);
|
||||
child.once("close", onClose);
|
||||
input.signal?.addEventListener("abort", onAbort, { once: true });
|
||||
if (input.signal?.aborted) onAbort();
|
||||
});
|
||||
}
|
||||
|
||||
function abortedError(signal: AbortSignal | undefined, label: string): AgentRunError {
|
||||
const reason = signal?.reason instanceof Error ? signal.reason.message : typeof signal?.reason === "string" ? signal.reason : "dispatch attempt deadline exceeded";
|
||||
return new AgentRunError("infra-failed", `${label} aborted: ${redactText(reason)}`, { httpStatus: 504, details: { aborted: true, valuesPrinted: false } });
|
||||
}
|
||||
@@ -14,17 +14,18 @@ export interface KafkaOutboxRelayOptions {
|
||||
}
|
||||
|
||||
export async function relayKafkaEventOutboxOnce(input: { store: AgentRunStore; options: KafkaOutboxRelayOptions; publish?: (config: AgentRunKafkaConfig, message: AgentRunKafkaMessage) => Promise<void> }): Promise<JsonRecord> {
|
||||
if (!input.options.enabled) return { action: "kafka-event-outbox-relay", enabled: false, claimedCount: 0, deliveredCount: 0, retryCount: 0, valuesPrinted: false };
|
||||
if (!input.options.enabled) return { action: "kafka-event-outbox-relay", enabled: false, claimedCount: 0, deliveredCount: 0, retryCount: 0, staleCount: 0, valuesPrinted: false };
|
||||
const claimed = await input.store.claimKafkaEventOutbox({ owner: input.options.owner, leaseMs: input.options.leaseMs, limit: input.options.batchSize });
|
||||
const items = claimed.slice().sort((a, b) => a.outboxSeq - b.outboxSeq);
|
||||
const blockedKeys = new Set<string>();
|
||||
let deliveredCount = 0;
|
||||
let retryCount = 0;
|
||||
let staleCount = 0;
|
||||
|
||||
for (const item of items) {
|
||||
if (blockedKeys.has(item.partitionKey)) {
|
||||
await retry(input.store, input.options, item, { code: "partition-key-blocked", message: "an earlier outbox record for this partition key failed", valuesPrinted: false });
|
||||
retryCount++;
|
||||
if (await retry(input.store, input.options, item, { code: "partition-key-blocked", message: "an earlier outbox record for this partition key failed", valuesPrinted: false })) retryCount++;
|
||||
else staleCount++;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
@@ -34,21 +35,22 @@ export async function relayKafkaEventOutboxOnce(input: { store: AgentRunStore; o
|
||||
value: item.value,
|
||||
headers: stringHeaders(item.headers),
|
||||
});
|
||||
await input.store.completeKafkaEventOutbox(item.id);
|
||||
await input.store.completeKafkaEventOutbox(item);
|
||||
deliveredCount++;
|
||||
} catch (error) {
|
||||
blockedKeys.add(item.partitionKey);
|
||||
await retry(input.store, input.options, item, {
|
||||
const retried = await retry(input.store, input.options, item, {
|
||||
code: "kafka-produce-failed",
|
||||
message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 500),
|
||||
topic: item.topic,
|
||||
outboxSeq: item.outboxSeq,
|
||||
valuesPrinted: false,
|
||||
});
|
||||
retryCount++;
|
||||
if (retried) retryCount++;
|
||||
else staleCount++;
|
||||
}
|
||||
}
|
||||
return { action: "kafka-event-outbox-relay", enabled: true, claimedCount: items.length, deliveredCount, retryCount, firstOutboxSeq: items[0]?.outboxSeq ?? null, lastOutboxSeq: items.at(-1)?.outboxSeq ?? null, valuesPrinted: false };
|
||||
return { action: "kafka-event-outbox-relay", enabled: true, claimedCount: items.length, deliveredCount, retryCount, staleCount, firstOutboxSeq: items[0]?.outboxSeq ?? null, lastOutboxSeq: items.at(-1)?.outboxSeq ?? null, valuesPrinted: false };
|
||||
}
|
||||
|
||||
export function startKafkaEventOutboxRelay(input: { store: AgentRunStore; options: KafkaOutboxRelayOptions; onError?: (error: unknown) => void }): () => Promise<void> {
|
||||
@@ -81,8 +83,14 @@ export function startKafkaEventOutboxRelay(input: { store: AgentRunStore; option
|
||||
};
|
||||
}
|
||||
|
||||
async function retry(store: AgentRunStore, options: KafkaOutboxRelayOptions, item: KafkaEventOutboxRecord, error: JsonRecord): Promise<void> {
|
||||
await store.retryKafkaEventOutbox(item.id, error, new Date(Date.now() + options.retryBackoffMs).toISOString());
|
||||
async function retry(store: AgentRunStore, options: KafkaOutboxRelayOptions, item: KafkaEventOutboxRecord, error: JsonRecord): Promise<boolean> {
|
||||
try {
|
||||
await store.retryKafkaEventOutbox(item, error, new Date(Date.now() + options.retryBackoffMs).toISOString());
|
||||
return true;
|
||||
} catch (settleError) {
|
||||
if (typeof settleError === "object" && settleError !== null && (settleError as { failureKind?: unknown }).failureKind === "runner-lease-conflict") return false;
|
||||
throw settleError;
|
||||
}
|
||||
}
|
||||
|
||||
function stringHeaders(headers: JsonRecord): Record<string, string> {
|
||||
|
||||
@@ -13,6 +13,7 @@ import { ensureSessionPvc } from "./session-pvc.js";
|
||||
import { gitTransportSummary } from "../common/git-transport.js";
|
||||
import { enforceRunnerRetentionBeforeCreate } from "./runner-retention.js";
|
||||
import type { RunnerRetentionOptions, RunnerRetentionSummary } from "./runner-retention.js";
|
||||
import { throwIfAborted, waitForChildProcess } from "./abortable-child.js";
|
||||
|
||||
const reusableCredentialEnvNames = new Set([
|
||||
"AUTH_PASSWORD",
|
||||
@@ -76,10 +77,12 @@ export interface CreateRunnerJobInput extends JsonRecord {
|
||||
transientEnv?: JsonRecord[];
|
||||
}
|
||||
|
||||
export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults }): Promise<JsonRecord> {
|
||||
export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise<JsonRecord> {
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
const commandId = stringField(options.input, "commandId");
|
||||
const run = await options.store.getRun(options.runId);
|
||||
const command = await options.store.getCommand(commandId);
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
if (command.runId !== run.id) throw new AgentRunError("schema-invalid", `command ${commandId} does not belong to run ${run.id}`, { httpStatus: 400 });
|
||||
if (command.type !== "turn") throw new AgentRunError("schema-invalid", `command ${commandId} is not a turn command`, { httpStatus: 400 });
|
||||
|
||||
@@ -90,6 +93,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
...(options.defaults.envIdentity ? { envIdentity: options.defaults.envIdentity } : {}),
|
||||
...(options.defaults.artifactCatalogFile ? { artifactCatalogFile: options.defaults.artifactCatalogFile } : {}),
|
||||
});
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
const image = envImage.image;
|
||||
const namespace = optionalString(options.input.namespace) ?? options.defaults.namespace;
|
||||
const managerUrl = optionalString(options.input.managerUrl) ?? options.defaults.managerUrl;
|
||||
@@ -128,11 +132,13 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
const payloadHash = stableHash(normalizedPayload);
|
||||
if (idempotencyKey) {
|
||||
const existing = await options.store.getRunnerJobByIdempotencyKey(run.id, idempotencyKey, payloadHash);
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
if (existing) return { ...existing.result, idempotentReplay: true };
|
||||
}
|
||||
if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${run.id} is already terminal: ${run.status}`, { httpStatus: 409 });
|
||||
if (isTerminalCommandState(command.state) || command.state !== "pending") throw new AgentRunError(command.state === "cancelled" ? "cancelled" : "schema-invalid", `command ${commandId} is not pending: ${command.state}`, { httpStatus: 409 });
|
||||
if (run.claimedBy && !isLeaseExpired(run.leaseExpiresAt)) {
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
const response = {
|
||||
action: "reuse-active-runner",
|
||||
mutation: false,
|
||||
@@ -150,16 +156,19 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
leaseExpiresAt: run.leaseExpiresAt,
|
||||
reason: "fresh-active-runner-lease",
|
||||
});
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
return response;
|
||||
}
|
||||
let preCreateRetention: RunnerRetentionSummary | null = null;
|
||||
if (options.defaults.retention) {
|
||||
preCreateRetention = await enforceRunnerRetentionBeforeCreate({ store: options.store, options: { ...options.defaults.retention, namespace }, incomingRunnerCount: 1 });
|
||||
preCreateRetention = await enforceRunnerRetentionBeforeCreate({ store: options.store, options: { ...options.defaults.retention, namespace, ...(options.signal ? { signal: options.signal } : {}) }, incomingRunnerCount: 1 });
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
}
|
||||
let sessionPvc: RunnerSessionPvcOptions | undefined;
|
||||
let sessionPvcSummary: JsonRecord | null = null;
|
||||
if (run.sessionRef?.sessionId) {
|
||||
const session = await options.store.getSession(run.sessionRef.sessionId);
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
if (session?.storageKind === "evicted") {
|
||||
throw new AgentRunError("session-store-evicted", `session ${session.sessionId} storage has been evicted; create a new sessionId`, { httpStatus: 409, details: { sessionId: session.sessionId, pvcName: session.storagePvcName ?? null, pvcPhase: session.storagePvcPhase ?? null, valuesPrinted: false } });
|
||||
}
|
||||
@@ -168,9 +177,11 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
store: options.store,
|
||||
sessionId: run.sessionRef.sessionId,
|
||||
namespace,
|
||||
options: { ...(options.defaults.kubectlCommand ? { kubectlCommand: options.defaults.kubectlCommand } : {}), defaultCodexRolloutSubdir: session.codexRolloutSubdir ?? "sessions" },
|
||||
options: { ...(options.defaults.kubectlCommand ? { kubectlCommand: options.defaults.kubectlCommand } : {}), defaultCodexRolloutSubdir: session.codexRolloutSubdir ?? "sessions", ...(options.signal ? { signal: options.signal } : {}) },
|
||||
});
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
const refreshed = await options.store.getSession(run.sessionRef.sessionId);
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
if (refreshed?.storageKind === "evicted") {
|
||||
throw new AgentRunError("session-store-evicted", `session ${refreshed.sessionId} storage has been evicted; create a new sessionId`, { httpStatus: 409, details: { sessionId: refreshed.sessionId, pvcName: refreshed.storagePvcName ?? null, pvcPhase: refreshed.storagePvcPhase ?? null, valuesPrinted: false } });
|
||||
}
|
||||
@@ -213,6 +224,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
...(sessionPvc ? { sessionPvc } : {}),
|
||||
};
|
||||
const render = renderRunnerJobManifest({ ...renderOptions, attemptId, ...(runnerId ? { runnerId } : {}), ...(runnerJobId ? { runnerJobId } : {}) });
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
const kubectlCommand = options.defaults.kubectlCommand ?? "kubectl";
|
||||
let transientEnvSecretCreated = false;
|
||||
let transientEnvSecretOwnerAttached = false;
|
||||
@@ -220,19 +232,20 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
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");
|
||||
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);
|
||||
transientEnvSecretCreated = !secretCreate.adopted;
|
||||
}
|
||||
const jobCreate = await kubectlCreateOrAdopt(render.manifest, kubectlCommand, "runner job");
|
||||
const jobCreate = await kubectlCreateOrAdopt(render.manifest, kubectlCommand, "runner job", options.signal);
|
||||
created = jobCreate.object;
|
||||
adopted = jobCreate.adopted;
|
||||
} catch (error) {
|
||||
if (transientEnvSecretName && transientEnvSecretCreated) await kubectlDeleteSecret(transientEnvSecretName, render.namespace, kubectlCommand);
|
||||
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 });
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
if (transientEnvSecretName) {
|
||||
const owner = await kubectlPatchSecretOwnerReference(transientEnvSecretName, render.namespace, { name: render.jobName, uid: objectPath(created, ["metadata", "uid"]) }, kubectlCommand);
|
||||
const owner = await kubectlPatchSecretOwnerReference(transientEnvSecretName, render.namespace, { name: render.jobName, uid: objectPath(created, ["metadata", "uid"]) }, kubectlCommand, options.signal);
|
||||
transientEnvSecretOwnerAttached = owner.ok;
|
||||
if (!owner.ok) render.warnings.push("transientEnv Secret ownerReference patch failed; Kubernetes TTL may not garbage-collect this per-job Secret automatically");
|
||||
}
|
||||
@@ -317,6 +330,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
serviceAccountName: serviceAccountName ?? null,
|
||||
result: response,
|
||||
});
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
await options.store.appendEvent(run.id, "backend_status", {
|
||||
phase: "runner-job-created",
|
||||
commandId,
|
||||
@@ -337,6 +351,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore;
|
||||
sessionRef: summarizeSessionRef(run.sessionRef ?? null),
|
||||
resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null),
|
||||
});
|
||||
throwIfAborted(options.signal, "runner dispatch attempt");
|
||||
return response;
|
||||
}
|
||||
|
||||
@@ -472,7 +487,8 @@ function summarizeTransientEnvSecret(name: string, namespace: string, items: Run
|
||||
};
|
||||
}
|
||||
|
||||
async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string, label = "runner job"): Promise<{ object: JsonRecord; adopted: boolean }> {
|
||||
async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string, label = "runner job", abortSignal?: AbortSignal): Promise<{ object: JsonRecord; adopted: boolean }> {
|
||||
throwIfAborted(abortSignal, `kubectl create ${label}`);
|
||||
const child = spawn(kubectlCommand, ["create", "-f", "-", "-o", "json"], { stdio: ["pipe", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
@@ -480,13 +496,10 @@ async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
|
||||
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
|
||||
const completion = waitForChildProcess(child, { ...(abortSignal ? { signal: abortSignal } : {}), label: `kubectl create ${label}` });
|
||||
child.stdin.on("error", () => {});
|
||||
child.stdin.end(`${JSON.stringify(manifest)}\n`);
|
||||
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 });
|
||||
});
|
||||
const result = await completion;
|
||||
if (result.code !== 0) {
|
||||
if (/alreadyexists|already exists/iu.test(stderr.replace(/\s+/gu, ""))) {
|
||||
const metadata = objectRecord(manifest.metadata);
|
||||
@@ -494,7 +507,7 @@ async function kubectlCreateOrAdopt(manifest: JsonRecord, kubectlCommand: string
|
||||
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 });
|
||||
const existing = await kubectlRun(kubectlCommand, ["get", kind, name, "-n", namespace, "-o", "json"]);
|
||||
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);
|
||||
@@ -550,22 +563,23 @@ function objectRecord(value: unknown): JsonRecord {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : {};
|
||||
}
|
||||
|
||||
async function kubectlDeleteSecret(name: string, namespace: string, kubectlCommand: string): Promise<void> {
|
||||
await kubectlRun(kubectlCommand, ["delete", "secret", name, "-n", namespace, "--ignore-not-found=true"]);
|
||||
async function kubectlDeleteSecret(name: string, namespace: string, kubectlCommand: string, abortSignal?: AbortSignal): Promise<void> {
|
||||
await kubectlRun(kubectlCommand, ["delete", "secret", name, "-n", namespace, "--ignore-not-found=true"], abortSignal);
|
||||
}
|
||||
|
||||
async function kubectlPatchSecretOwnerReference(name: string, namespace: string, owner: { name: string; uid: string | null }, kubectlCommand: string): Promise<{ ok: boolean }> {
|
||||
async function kubectlPatchSecretOwnerReference(name: string, namespace: string, owner: { name: string; uid: string | null }, kubectlCommand: string, abortSignal?: AbortSignal): Promise<{ ok: boolean }> {
|
||||
if (!owner.uid) return { ok: false };
|
||||
const patch = {
|
||||
metadata: {
|
||||
ownerReferences: [{ apiVersion: "batch/v1", kind: "Job", name: owner.name, uid: owner.uid, controller: false, blockOwnerDeletion: false }],
|
||||
},
|
||||
};
|
||||
const result = await kubectlRun(kubectlCommand, ["patch", "secret", name, "-n", namespace, "--type", "merge", "-p", JSON.stringify(patch), "-o", "json"]);
|
||||
const result = await kubectlRun(kubectlCommand, ["patch", "secret", name, "-n", namespace, "--type", "merge", "-p", JSON.stringify(patch), "-o", "json"], abortSignal);
|
||||
return { ok: result.code === 0 };
|
||||
}
|
||||
|
||||
async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
||||
async function kubectlRun(kubectlCommand: string, args: string[], abortSignal?: AbortSignal): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
||||
throwIfAborted(abortSignal, "kubectl");
|
||||
const child = spawn(kubectlCommand, args, { stdio: ["ignore", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
@@ -573,12 +587,7 @@ async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ cod
|
||||
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 });
|
||||
});
|
||||
const result = await waitForChildProcess(child, { ...(abortSignal ? { signal: abortSignal } : {}), label: "kubectl" });
|
||||
return { ...result, stdout, stderr };
|
||||
}
|
||||
|
||||
|
||||
@@ -51,6 +51,12 @@ CREATE INDEX IF NOT EXISTS agentrun_kafka_event_outbox_due_idx
|
||||
ON agentrun_kafka_event_outbox (delivered_at, next_attempt_at, lease_expires_at);
|
||||
`;
|
||||
|
||||
export const runnerDispatchOutcomeMigrationSql = `
|
||||
ALTER TABLE agentrun_runner_dispatch_intents ADD COLUMN IF NOT EXISTS dispatch_outcome text;
|
||||
ALTER TABLE agentrun_runner_dispatch_intents ADD COLUMN IF NOT EXISTS actual_runner_job_id text;
|
||||
ALTER TABLE agentrun_runner_dispatch_intents ADD COLUMN IF NOT EXISTS active_runner_id text;
|
||||
`;
|
||||
|
||||
export function runnerDispatchIntentFromRow(row: QueryResultRow): RunnerDispatchIntentRecord {
|
||||
return {
|
||||
id: stringValue(row.id),
|
||||
@@ -62,6 +68,9 @@ export function runnerDispatchIntentFromRow(row: QueryResultRow): RunnerDispatch
|
||||
state: stringValue(row.state) as RunnerDispatchIntentRecord["state"],
|
||||
attemptCount: Number(row.attempt_count ?? 0),
|
||||
runnerJobId: stringValue(row.runner_job_id),
|
||||
dispatchOutcome: nullableString(row.dispatch_outcome) as RunnerDispatchIntentRecord["dispatchOutcome"],
|
||||
actualRunnerJobId: nullableString(row.actual_runner_job_id),
|
||||
activeRunnerId: nullableString(row.active_runner_id),
|
||||
leaseOwner: nullableString(row.lease_owner),
|
||||
leaseExpiresAt: nullableIso(row.lease_expires_at),
|
||||
nextAttemptAt: iso(row.next_attempt_at),
|
||||
|
||||
+61
-36
@@ -3,15 +3,15 @@ import { Pool } from "pg";
|
||||
import type { PoolClient, QueryResultRow } from "pg";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactJson } from "../common/redaction.js";
|
||||
import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
|
||||
import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, 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 { AgentRunStore, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
|
||||
import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
|
||||
import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, lateWriteRejectedPayload, parseQueueCursor, parseSessionCursor, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
|
||||
import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
|
||||
import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
|
||||
import { buildKafkaEventOutboxRecord } from "./event-outbox.js";
|
||||
import { assertRunnerDispatchReplay, attachRunnerDispatchIntent, newRunnerDispatchIntent } from "./runner-dispatch-intent.js";
|
||||
import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, runnerDispatchIntentFromRow } from "./postgres-durable-queues.js";
|
||||
import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, runnerDispatchIntentFromRow, runnerDispatchOutcomeMigrationSql } from "./postgres-durable-queues.js";
|
||||
|
||||
interface PostgresStoreOptions {
|
||||
connectionString: string;
|
||||
@@ -433,6 +433,11 @@ const postgresMigrations: MigrationDefinition[] = [
|
||||
checksum: checksumSql(durableDispatchAndKafkaOutboxMigrationSql),
|
||||
sql: durableDispatchAndKafkaOutboxMigrationSql,
|
||||
},
|
||||
{
|
||||
id: "013_v02_runner_dispatch_outcome",
|
||||
checksum: checksumSql(runnerDispatchOutcomeMigrationSql),
|
||||
sql: runnerDispatchOutcomeMigrationSql,
|
||||
},
|
||||
];
|
||||
|
||||
export function postgresMigrationContract(): JsonRecord {
|
||||
@@ -621,32 +626,46 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return result.rows.map(runnerDispatchIntentFromRow);
|
||||
}
|
||||
|
||||
async completeRunnerDispatchIntent(intentId: string, runnerJobId: string): Promise<RunnerDispatchIntentRecord> {
|
||||
const result = await this.pool.query(
|
||||
`UPDATE agentrun_runner_dispatch_intents
|
||||
SET state = 'dispatched', runner_job_id = $2, lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, dispatched_at = now(), updated_at = now()
|
||||
WHERE id = $1 RETURNING *`,
|
||||
[intentId, runnerJobId],
|
||||
);
|
||||
if (!result.rows[0]) throw new AgentRunError("schema-invalid", `runner dispatch intent ${intentId} was not found`, { httpStatus: 404 });
|
||||
return runnerDispatchIntentFromRow(result.rows[0]);
|
||||
}
|
||||
|
||||
async retryRunnerDispatchIntent(intentId: string, error: JsonRecord, nextAttemptAt: string): Promise<RunnerDispatchIntentRecord> {
|
||||
const result = await this.pool.query(
|
||||
`UPDATE agentrun_runner_dispatch_intents
|
||||
SET state = 'retry', lease_owner = NULL, lease_expires_at = NULL, next_attempt_at = $2, last_error = $3::jsonb, updated_at = now()
|
||||
WHERE id = $1 RETURNING *`,
|
||||
[intentId, nextAttemptAt, JSON.stringify(redactJson(error))],
|
||||
);
|
||||
if (!result.rows[0]) throw new AgentRunError("schema-invalid", `runner dispatch intent ${intentId} was not found`, { httpStatus: 404 });
|
||||
return runnerDispatchIntentFromRow(result.rows[0]);
|
||||
}
|
||||
|
||||
async terminalizeRunnerDispatchIntent(intentId: string, error: JsonRecord): Promise<JsonRecord> {
|
||||
async completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): Promise<RunnerDispatchIntentRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const intentResult = await client.query("SELECT * FROM agentrun_runner_dispatch_intents WHERE id = $1 FOR UPDATE", [intentId]);
|
||||
if (!intentResult.rows[0]) throw new AgentRunError("schema-invalid", `runner dispatch intent ${intentId} was not found`, { httpStatus: 404 });
|
||||
const result = await client.query(
|
||||
`UPDATE agentrun_runner_dispatch_intents
|
||||
SET state = 'dispatched', dispatch_outcome = $2, actual_runner_job_id = $3, active_runner_id = $4,
|
||||
lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, dispatched_at = now(), updated_at = now()
|
||||
WHERE id = $1 AND state = 'dispatching' AND lease_owner = $5 AND attempt_count = $6 AND lease_expires_at > now()
|
||||
RETURNING *`,
|
||||
[claim.id, completion.dispatchOutcome, completion.actualRunnerJobId, completion.activeRunnerId, claim.leaseOwner, claim.attemptCount],
|
||||
);
|
||||
if (!result.rows[0]) throw staleClaimError("runner dispatch", claim);
|
||||
const intent = runnerDispatchIntentFromRow(result.rows[0]);
|
||||
await this.appendEventWithLockedRun(client, intent.runId, "backend_status", eventPayload);
|
||||
return intent;
|
||||
});
|
||||
}
|
||||
|
||||
async retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): Promise<RunnerDispatchIntentRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const result = await client.query(
|
||||
`UPDATE agentrun_runner_dispatch_intents
|
||||
SET state = 'retry', lease_owner = NULL, lease_expires_at = NULL, next_attempt_at = $2, last_error = $3::jsonb, updated_at = now()
|
||||
WHERE id = $1 AND state = 'dispatching' AND lease_owner = $4 AND attempt_count = $5 AND lease_expires_at > now()
|
||||
RETURNING *`,
|
||||
[claim.id, nextAttemptAt, JSON.stringify(redactJson(error)), claim.leaseOwner, claim.attemptCount],
|
||||
);
|
||||
if (!result.rows[0]) throw staleClaimError("runner dispatch", claim);
|
||||
const intent = runnerDispatchIntentFromRow(result.rows[0]);
|
||||
await this.appendEventWithLockedRun(client, intent.runId, "backend_status", eventPayload);
|
||||
return intent;
|
||||
});
|
||||
}
|
||||
|
||||
async terminalizeRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord): Promise<JsonRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const intentResult = await client.query(
|
||||
"SELECT * FROM agentrun_runner_dispatch_intents WHERE id = $1 AND state = 'dispatching' AND lease_owner = $2 AND attempt_count = $3 AND lease_expires_at > now() FOR UPDATE",
|
||||
[claim.id, claim.leaseOwner, claim.attemptCount],
|
||||
);
|
||||
if (!intentResult.rows[0]) throw staleClaimError("runner dispatch", claim);
|
||||
const intent = runnerDispatchIntentFromRow(intentResult.rows[0]);
|
||||
const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [intent.commandId]);
|
||||
if (!commandResult.rows[0]) throw new AgentRunError("schema-invalid", `command ${intent.commandId} was not found`, { httpStatus: 404 });
|
||||
@@ -724,25 +743,27 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return result.rows.map(kafkaEventOutboxFromRow);
|
||||
}
|
||||
|
||||
async completeKafkaEventOutbox(outboxId: string): Promise<KafkaEventOutboxRecord> {
|
||||
async completeKafkaEventOutbox(claim: DurableQueueClaim): Promise<KafkaEventOutboxRecord> {
|
||||
const result = await this.pool.query(
|
||||
`UPDATE agentrun_kafka_event_outbox
|
||||
SET delivered_at = now(), lease_owner = NULL, lease_expires_at = NULL, last_error = NULL, updated_at = now()
|
||||
WHERE id = $1 RETURNING *`,
|
||||
[outboxId],
|
||||
WHERE id = $1 AND delivered_at IS NULL AND lease_owner = $2 AND attempt_count = $3 AND lease_expires_at > now()
|
||||
RETURNING *`,
|
||||
[claim.id, claim.leaseOwner, claim.attemptCount],
|
||||
);
|
||||
if (!result.rows[0]) throw new AgentRunError("schema-invalid", `Kafka event outbox ${outboxId} was not found`, { httpStatus: 404 });
|
||||
if (!result.rows[0]) throw staleClaimError("Kafka event outbox", claim);
|
||||
return kafkaEventOutboxFromRow(result.rows[0]);
|
||||
}
|
||||
|
||||
async retryKafkaEventOutbox(outboxId: string, error: JsonRecord, nextAttemptAt: string): Promise<KafkaEventOutboxRecord> {
|
||||
async retryKafkaEventOutbox(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string): Promise<KafkaEventOutboxRecord> {
|
||||
const result = await this.pool.query(
|
||||
`UPDATE agentrun_kafka_event_outbox
|
||||
SET lease_owner = NULL, lease_expires_at = NULL, next_attempt_at = $2, last_error = $3::jsonb, updated_at = now()
|
||||
WHERE id = $1 RETURNING *`,
|
||||
[outboxId, nextAttemptAt, JSON.stringify(redactJson(error))],
|
||||
WHERE id = $1 AND delivered_at IS NULL AND lease_owner = $4 AND attempt_count = $5 AND lease_expires_at > now()
|
||||
RETURNING *`,
|
||||
[claim.id, nextAttemptAt, JSON.stringify(redactJson(error)), claim.leaseOwner, claim.attemptCount],
|
||||
);
|
||||
if (!result.rows[0]) throw new AgentRunError("schema-invalid", `Kafka event outbox ${outboxId} was not found`, { httpStatus: 404 });
|
||||
if (!result.rows[0]) throw staleClaimError("Kafka event outbox", claim);
|
||||
return kafkaEventOutboxFromRow(result.rows[0]);
|
||||
}
|
||||
|
||||
@@ -1654,6 +1675,10 @@ function clamp(value: number, min: number, max: number): number {
|
||||
return Math.max(min, Math.min(value, max));
|
||||
}
|
||||
|
||||
function staleClaimError(component: string, claim: DurableQueueClaim): AgentRunError {
|
||||
return new AgentRunError("runner-lease-conflict", `${component} claim is stale`, { httpStatus: 409, details: { id: claim.id, attemptCount: claim.attemptCount, valuesPrinted: false } });
|
||||
}
|
||||
|
||||
function runFromRow(row: QueryResultRow): RunRecord {
|
||||
return {
|
||||
id: stringValue(row.id),
|
||||
|
||||
@@ -24,6 +24,9 @@ export function newRunnerDispatchIntent(command: CommandRecord, request: RunnerD
|
||||
state: "pending",
|
||||
attemptCount: 0,
|
||||
runnerJobId,
|
||||
dispatchOutcome: null,
|
||||
actualRunnerJobId: null,
|
||||
activeRunnerId: null,
|
||||
leaseOwner: null,
|
||||
leaseExpiresAt: null,
|
||||
nextAttemptAt: at,
|
||||
@@ -43,6 +46,9 @@ export function attachRunnerDispatchIntent(command: CommandRecord, intent: Runne
|
||||
id: intent.id,
|
||||
state: intent.state,
|
||||
runnerJobId: intent.runnerJobId,
|
||||
dispatchOutcome: intent.dispatchOutcome,
|
||||
actualRunnerJobId: intent.actualRunnerJobId,
|
||||
activeRunnerId: intent.activeRunnerId,
|
||||
attemptCount: intent.attemptCount,
|
||||
durable: true,
|
||||
} : null,
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
import type { JsonRecord, RunnerDispatchIntentRecord } from "../common/types.js";
|
||||
import type { JsonRecord, RunnerDispatchCompletion, RunnerDispatchIntentRecord } from "../common/types.js";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactText } from "../common/redaction.js";
|
||||
import { nowIso } from "../common/validation.js";
|
||||
import type { AgentRunStore } from "./store.js";
|
||||
import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js";
|
||||
import { childProcessKillGraceMs } from "./abortable-child.js";
|
||||
|
||||
export interface RunnerDispatcherOptions {
|
||||
enabled: boolean;
|
||||
@@ -11,6 +13,7 @@ export interface RunnerDispatcherOptions {
|
||||
leaseMs: number;
|
||||
maxAttempts: number;
|
||||
retryBackoffMs: number;
|
||||
attemptTimeoutMs: number;
|
||||
owner: string;
|
||||
}
|
||||
|
||||
@@ -18,38 +21,55 @@ export async function dispatchRunnerIntentsOnce(input: {
|
||||
store: AgentRunStore;
|
||||
defaults: RunnerJobDefaults;
|
||||
options: RunnerDispatcherOptions;
|
||||
signal?: AbortSignal;
|
||||
}): Promise<JsonRecord> {
|
||||
if (!input.options.enabled) return { action: "runner-dispatch", enabled: false, claimedCount: 0, dispatchedCount: 0, retryCount: 0, failedCount: 0, valuesPrinted: false };
|
||||
const intents = await input.store.claimRunnerDispatchIntents({ owner: input.options.owner, leaseMs: input.options.leaseMs, limit: input.options.batchSize });
|
||||
if (!input.options.enabled) return { action: "runner-dispatch", enabled: false, claimedCount: 0, dispatchedCount: 0, retryCount: 0, failedCount: 0, staleCount: 0, valuesPrinted: false };
|
||||
assertRunnerDispatcherAttemptTiming(input.options);
|
||||
const items: JsonRecord[] = [];
|
||||
let claimedCount = 0;
|
||||
let dispatchedCount = 0;
|
||||
let retryCount = 0;
|
||||
let failedCount = 0;
|
||||
let staleCount = 0;
|
||||
|
||||
for (const intent of intents) {
|
||||
for (let index = 0; index < input.options.batchSize; index++) {
|
||||
if (input.signal?.aborted) break;
|
||||
const intent = (await input.store.claimRunnerDispatchIntents({ owner: input.options.owner, leaseMs: input.options.leaseMs, limit: 1 }))[0];
|
||||
if (!intent) break;
|
||||
claimedCount++;
|
||||
try {
|
||||
const result = await createKubernetesRunnerJob({ store: input.store, runId: intent.runId, input: intent.input as never, defaults: input.defaults });
|
||||
const runnerJobId = stringValue(result.runnerJobId) ?? intent.runnerJobId;
|
||||
await input.store.completeRunnerDispatchIntent(intent.id, runnerJobId);
|
||||
await input.store.appendEvent(intent.runId, "backend_status", { phase: "runner-dispatch-completed", commandId: intent.commandId, dispatchIntentId: intent.id, runnerJobId, attemptCount: intent.attemptCount, adopted: result.adopted === true, valuesPrinted: false });
|
||||
const result = await withAttemptDeadline(intent, input.options.attemptTimeoutMs, input.signal, async (signal) => await createKubernetesRunnerJob({ store: input.store, runId: intent.runId, input: intent.input as never, defaults: input.defaults, signal }));
|
||||
const completion = completionFromResult(result);
|
||||
await input.store.completeRunnerDispatchIntent(intent, completion, { phase: "runner-dispatch-completed", commandId: intent.commandId, dispatchIntentId: intent.id, plannedRunnerJobId: intent.runnerJobId, runnerJobId: completion.actualRunnerJobId, activeRunnerId: completion.activeRunnerId, dispatchOutcome: completion.dispatchOutcome, attemptCount: intent.attemptCount, adopted: result.adopted === true, valuesPrinted: false });
|
||||
dispatchedCount++;
|
||||
items.push(itemSummary(intent, "dispatched", null));
|
||||
items.push(itemSummary(intent, "dispatched", null, completion));
|
||||
} catch (error) {
|
||||
if (!await currentClaim(input.store, intent)) {
|
||||
staleCount++;
|
||||
items.push(itemSummary(intent, "stale", null));
|
||||
continue;
|
||||
}
|
||||
const terminal = intent.attemptCount >= input.options.maxAttempts;
|
||||
const failure = dispatchFailure(error, intent);
|
||||
const nextAttemptAt = new Date(Date.now() + input.options.retryBackoffMs).toISOString();
|
||||
if (terminal) {
|
||||
await input.store.terminalizeRunnerDispatchIntent(intent.id, failure);
|
||||
failedCount++;
|
||||
} else {
|
||||
await input.store.retryRunnerDispatchIntent(intent.id, failure, nextAttemptAt);
|
||||
await input.store.appendEvent(intent.runId, "backend_status", { phase: "runner-dispatch-retry", commandId: intent.commandId, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, nextAttemptAt, failureKind: "infra-failed", message: failure.message, valuesPrinted: false });
|
||||
retryCount++;
|
||||
try {
|
||||
if (terminal) {
|
||||
await input.store.terminalizeRunnerDispatchIntent(intent, failure);
|
||||
failedCount++;
|
||||
} else {
|
||||
await input.store.retryRunnerDispatchIntent(intent, failure, nextAttemptAt, { phase: "runner-dispatch-retry", commandId: intent.commandId, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, nextAttemptAt, failureKind: "infra-failed", message: failure.message, valuesPrinted: false });
|
||||
retryCount++;
|
||||
}
|
||||
} catch (settleError) {
|
||||
if (!isLeaseConflict(settleError)) throw settleError;
|
||||
staleCount++;
|
||||
items.push(itemSummary(intent, "stale", null));
|
||||
continue;
|
||||
}
|
||||
items.push(itemSummary(intent, terminal ? "failed" : "retry", failure));
|
||||
}
|
||||
}
|
||||
return { action: "runner-dispatch", enabled: true, claimedCount: intents.length, dispatchedCount, retryCount, failedCount, items, valuesPrinted: false };
|
||||
return { action: "runner-dispatch", enabled: true, claimedCount, dispatchedCount, retryCount, failedCount, staleCount, items, valuesPrinted: false };
|
||||
}
|
||||
|
||||
export function startRunnerDispatcher(input: {
|
||||
@@ -60,11 +80,12 @@ export function startRunnerDispatcher(input: {
|
||||
}): () => void {
|
||||
let stopped = false;
|
||||
let running = false;
|
||||
const shutdown = new AbortController();
|
||||
const cycle = async (): Promise<void> => {
|
||||
if (stopped || running) return;
|
||||
running = true;
|
||||
try {
|
||||
await dispatchRunnerIntentsOnce(input);
|
||||
await dispatchRunnerIntentsOnce({ ...input, signal: shutdown.signal });
|
||||
} catch (error) {
|
||||
input.onError?.(error);
|
||||
} finally {
|
||||
@@ -74,7 +95,7 @@ export function startRunnerDispatcher(input: {
|
||||
void cycle();
|
||||
const timer = setInterval(() => { void cycle(); }, input.options.intervalMs);
|
||||
timer.unref?.();
|
||||
return () => { stopped = true; clearInterval(timer); };
|
||||
return () => { stopped = true; clearInterval(timer); shutdown.abort(new Error("runner dispatcher stopped")); };
|
||||
}
|
||||
|
||||
function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): JsonRecord {
|
||||
@@ -90,10 +111,55 @@ function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): Js
|
||||
};
|
||||
}
|
||||
|
||||
function itemSummary(intent: RunnerDispatchIntentRecord, state: string, error: JsonRecord | null): JsonRecord {
|
||||
return { dispatchIntentId: intent.id, runId: intent.runId, commandId: intent.commandId, runnerJobId: intent.runnerJobId, attemptCount: intent.attemptCount, state, failureKind: error?.failureKind ?? null, valuesPrinted: false };
|
||||
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 };
|
||||
}
|
||||
|
||||
async function currentClaim(store: AgentRunStore, claim: RunnerDispatchIntentRecord): Promise<boolean> {
|
||||
const current = await store.getRunnerDispatchIntent(claim.commandId);
|
||||
return current?.state === "dispatching"
|
||||
&& current.leaseOwner === claim.leaseOwner
|
||||
&& current.attemptCount === claim.attemptCount
|
||||
&& Date.parse(current.leaseExpiresAt ?? "") > Date.now();
|
||||
}
|
||||
|
||||
function isLeaseConflict(error: unknown): boolean {
|
||||
return typeof error === "object" && error !== null && (error as { failureKind?: unknown }).failureKind === "runner-lease-conflict";
|
||||
}
|
||||
|
||||
function stringValue(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
||||
}
|
||||
|
||||
function completionFromResult(result: JsonRecord): RunnerDispatchCompletion {
|
||||
if (result.action === "reuse-active-runner") {
|
||||
const activeRunnerId = stringValue(result.runnerId);
|
||||
if (!activeRunnerId) throw new AgentRunError("infra-failed", "active runner reuse did not return runnerId", { httpStatus: 502 });
|
||||
return { dispatchOutcome: "active-runner-reused", actualRunnerJobId: null, activeRunnerId };
|
||||
}
|
||||
if (result.action !== "create-kubernetes-job") throw new AgentRunError("infra-failed", "runner dispatch returned an unsupported outcome", { httpStatus: 502 });
|
||||
const actualRunnerJobId = stringValue(result.runnerJobId);
|
||||
if (!actualRunnerJobId) throw new AgentRunError("infra-failed", "Kubernetes runner dispatch did not return runnerJobId", { httpStatus: 502 });
|
||||
return { dispatchOutcome: "kubernetes-job", actualRunnerJobId, activeRunnerId: null };
|
||||
}
|
||||
|
||||
export function assertRunnerDispatcherAttemptTiming(options: Pick<RunnerDispatcherOptions, "attemptTimeoutMs" | "leaseMs">): void {
|
||||
if (options.attemptTimeoutMs + childProcessKillGraceMs >= options.leaseMs) {
|
||||
throw new AgentRunError("infra-failed", "runner dispatch attempt timeout plus child termination grace must be strictly less than its lease", { httpStatus: 503, details: { attemptTimeoutMs: options.attemptTimeoutMs, childTerminationGraceMs: childProcessKillGraceMs, leaseMs: options.leaseMs, valuesPrinted: false } });
|
||||
}
|
||||
}
|
||||
|
||||
async function withAttemptDeadline<T>(intent: RunnerDispatchIntentRecord, timeoutMs: number, parentSignal: AbortSignal | undefined, callback: (signal: AbortSignal) => Promise<T>): Promise<T> {
|
||||
const controller = new AbortController();
|
||||
const onParentAbort = (): void => controller.abort(parentSignal?.reason ?? new Error("runner dispatcher stopped"));
|
||||
parentSignal?.addEventListener("abort", onParentAbort, { once: true });
|
||||
if (parentSignal?.aborted) onParentAbort();
|
||||
const timer = setTimeout(() => controller.abort(new Error(`runner dispatch intent ${intent.id} exceeded ${timeoutMs}ms attempt timeout`)), timeoutMs);
|
||||
timer.unref?.();
|
||||
try {
|
||||
return await callback(controller.signal);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
parentSignal?.removeEventListener("abort", onParentAbort);
|
||||
}
|
||||
}
|
||||
|
||||
+13
-12
@@ -5,6 +5,7 @@ import type { CommandRecord, JsonRecord, JsonValue, RunRecord } from "../common/
|
||||
import { stableHash } from "../common/validation.js";
|
||||
import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
|
||||
import type { AgentRunStore } from "./store.js";
|
||||
import { throwIfAborted, waitForChildProcess } from "./abortable-child.js";
|
||||
|
||||
export interface RunnerRetentionOptions {
|
||||
namespace: string;
|
||||
@@ -18,6 +19,7 @@ export interface RunnerRetentionOptions {
|
||||
maxAgeHours: number | null;
|
||||
};
|
||||
kubectlCommand?: string;
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
|
||||
export interface RunnerRetentionSummary extends JsonRecord {
|
||||
@@ -118,11 +120,11 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
|
||||
let deletedAssociatedResourceCount = 0;
|
||||
for (const item of selected) {
|
||||
if (item.resourceKind === "job") {
|
||||
await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "job", item.name, "-n", input.options.namespace, "--ignore-not-found=true"]);
|
||||
await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "job", item.name, "-n", input.options.namespace, "--ignore-not-found=true"], input.options.signal);
|
||||
deletedRunnerJobCount++;
|
||||
deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
|
||||
} else {
|
||||
await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "pod", item.name, "-n", input.options.namespace, "--ignore-not-found=true"]);
|
||||
await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "pod", item.name, "-n", input.options.namespace, "--ignore-not-found=true"], input.options.signal);
|
||||
deletedRunnerPodCount++;
|
||||
}
|
||||
}
|
||||
@@ -153,10 +155,13 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
|
||||
}
|
||||
|
||||
async function runnerSnapshot(store: AgentRunStore, options: RunnerRetentionOptions): Promise<Snapshot> {
|
||||
const [jobObjects, podObjects] = await Promise.all([
|
||||
const observed = await Promise.allSettled([
|
||||
kubectlGetList(options, "jobs", labelSelector(options.matchLabels)),
|
||||
kubectlGetList(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 jobs: RunnerResourceEntry[] = [];
|
||||
for (const item of jobObjects) {
|
||||
const name = objectName(item);
|
||||
@@ -262,7 +267,7 @@ async function kubectlGetList(options: RunnerRetentionOptions, resource: string,
|
||||
const args = ["get", resource, "-n", options.namespace];
|
||||
if (selector) args.push("-l", selector);
|
||||
args.push("-o", "json");
|
||||
const result = await kubectlRun(options.kubectlCommand ?? "kubectl", args);
|
||||
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}`);
|
||||
const items = parsed.items;
|
||||
@@ -273,14 +278,15 @@ async function deleteAssociatedResources(options: RunnerRetentionOptions, jobNam
|
||||
let deleted = 0;
|
||||
const selector = `agentrun.pikastech.local/runner-job=${jobName}`;
|
||||
for (const resource of ["secret", "pvc"] as const) {
|
||||
const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", resource, "-n", options.namespace, "-l", selector, "--ignore-not-found=true"]);
|
||||
const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", resource, "-n", options.namespace, "-l", selector, "--ignore-not-found=true"], options.signal);
|
||||
if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl delete associated ${resource} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) });
|
||||
deleted += countDeletedLines(result.stdout);
|
||||
}
|
||||
return deleted;
|
||||
}
|
||||
|
||||
async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
||||
async function kubectlRun(kubectlCommand: string, args: string[], abortSignal?: AbortSignal): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
||||
throwIfAborted(abortSignal, "kubectl");
|
||||
const child = spawn(kubectlCommand, args, { stdio: ["ignore", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
@@ -288,12 +294,7 @@ async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ cod
|
||||
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 });
|
||||
});
|
||||
const result = await waitForChildProcess(child, { ...(abortSignal ? { signal: abortSignal } : {}), label: "kubectl" });
|
||||
return { ...result, stdout, stderr };
|
||||
}
|
||||
|
||||
|
||||
+23
-3
@@ -14,7 +14,7 @@ import { dispatchQueueTask, refreshQueueTaskFromCore } from "./queue-dispatch.js
|
||||
import { buildRunResult } from "./result.js";
|
||||
import { runnerJobStatusSummary } from "./runner-job-status.js";
|
||||
import { reconcileRunnerJobsOnce, startRunnerJobReconciler } from "./runner-reconciler.js";
|
||||
import { dispatchRunnerIntentsOnce, startRunnerDispatcher, type RunnerDispatcherOptions } from "./runner-dispatcher.js";
|
||||
import { assertRunnerDispatcherAttemptTiming, dispatchRunnerIntentsOnce, startRunnerDispatcher, type RunnerDispatcherOptions } from "./runner-dispatcher.js";
|
||||
import { relayKafkaEventOutboxOnce, startKafkaEventOutboxRelay, type KafkaOutboxRelayOptions } from "./kafka-outbox-relay.js";
|
||||
import { createSessionPvc, deleteSessionPvc, getSessionPvcSummary, refreshSessionPvcSummary, runSessionStorageGc } from "./session-pvc.js";
|
||||
import type { SessionPvcSummary } from "./session-pvc.js";
|
||||
@@ -68,6 +68,22 @@ function runnerJobDefaultsForRequest(defaults: ManagerServerOptions["runnerJobDe
|
||||
};
|
||||
}
|
||||
|
||||
function assertRunnerDispatcherRuntimeConfigured(enabled: boolean, defaults: ManagerServerOptions["runnerJobDefaults"]): void {
|
||||
if (!enabled) return;
|
||||
const required = [
|
||||
["AGENTRUN_RUNTIME_NAMESPACE", defaults?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE],
|
||||
["AGENTRUN_INTERNAL_MGR_URL", defaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL],
|
||||
["AGENTRUN_RUNNER_IMAGE", defaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE],
|
||||
["AGENTRUN_RUNNER_SERVICE_ACCOUNT", defaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT],
|
||||
["AGENTRUN_RUNNER_JOB_NAME_PREFIX", defaults?.jobNamePrefix ?? process.env.AGENTRUN_RUNNER_JOB_NAME_PREFIX],
|
||||
["AGENTRUN_LANE", defaults?.lane ?? process.env.AGENTRUN_LANE],
|
||||
] as const;
|
||||
const missing = required.filter(([, value]) => typeof value !== "string" || value.trim().length === 0).map(([name]) => name);
|
||||
if (missing.length > 0) {
|
||||
throw new AgentRunError("infra-failed", "durable runner dispatcher runtime configuration is incomplete", { httpStatus: 503, details: { missing, valuesPrinted: false } });
|
||||
}
|
||||
}
|
||||
|
||||
function runnerRetentionOptionsForRequest(defaults: RunnerRetentionOptions | undefined, namespace: string, jobNamePrefix: string, kubectlCommand?: string): RunnerRetentionOptions | undefined {
|
||||
if (defaults) return { ...defaults, namespace, ...(kubectlCommand ? { kubectlCommand } : {}) };
|
||||
const maxRunners = optionalPositiveInteger("AGENTRUN_RUNNER_RETENTION_MAX_RUNNERS", process.env.AGENTRUN_RUNNER_RETENTION_MAX_RUNNERS);
|
||||
@@ -102,15 +118,18 @@ function runnerReconcilerOptionsForRuntime(defaults: ManagerServerOptions["runne
|
||||
|
||||
function runnerDispatcherOptionsForRuntime(defaults: ManagerServerOptions["runnerDispatcherOptions"] | undefined): RunnerDispatcherOptions {
|
||||
const enabled = defaults?.enabled ?? optionalBooleanEnv("AGENTRUN_RUNNER_DISPATCHER_ENABLED", process.env.AGENTRUN_RUNNER_DISPATCHER_ENABLED, false);
|
||||
return {
|
||||
const options: RunnerDispatcherOptions = {
|
||||
enabled,
|
||||
intervalMs: runtimePositiveInteger(defaults?.intervalMs, "AGENTRUN_RUNNER_DISPATCHER_INTERVAL_MS", enabled),
|
||||
batchSize: runtimePositiveInteger(defaults?.batchSize, "AGENTRUN_RUNNER_DISPATCHER_BATCH_SIZE", enabled),
|
||||
leaseMs: runtimePositiveInteger(defaults?.leaseMs, "AGENTRUN_RUNNER_DISPATCHER_LEASE_MS", enabled),
|
||||
maxAttempts: runtimePositiveInteger(defaults?.maxAttempts, "AGENTRUN_RUNNER_DISPATCHER_MAX_ATTEMPTS", enabled),
|
||||
retryBackoffMs: runtimePositiveInteger(defaults?.retryBackoffMs, "AGENTRUN_RUNNER_DISPATCHER_RETRY_BACKOFF_MS", enabled),
|
||||
attemptTimeoutMs: runtimePositiveInteger(defaults?.attemptTimeoutMs, "AGENTRUN_RUNNER_DISPATCHER_ATTEMPT_TIMEOUT_MS", enabled),
|
||||
owner: defaults?.owner ?? process.env.HOSTNAME ?? `agentrun-mgr-${process.pid}`,
|
||||
};
|
||||
if (enabled) assertRunnerDispatcherAttemptTiming(options);
|
||||
return options;
|
||||
}
|
||||
|
||||
function kafkaOutboxRelayOptionsForRuntime(defaults: ManagerServerOptions["kafkaOutboxRelayOptions"] | undefined): KafkaOutboxRelayOptions {
|
||||
@@ -162,7 +181,7 @@ export interface ManagerServerOptions {
|
||||
providerProfileOptions?: { namespace?: string; kubectlCommand?: string };
|
||||
toolCredentialOptions?: { namespace?: string; kubectlCommand?: string };
|
||||
runnerReconcilerOptions?: { enabled?: boolean; namespace?: string; kubectlCommand?: string; intervalMs?: number; batchSize?: number };
|
||||
runnerDispatcherOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; maxAttempts?: number; retryBackoffMs?: number; owner?: string };
|
||||
runnerDispatcherOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; maxAttempts?: number; retryBackoffMs?: number; attemptTimeoutMs?: number; owner?: string };
|
||||
kafkaOutboxRelayOptions?: { enabled?: boolean; intervalMs?: number; batchSize?: number; leaseMs?: number; retryBackoffMs?: number; owner?: string; config?: AgentRunKafkaConfig };
|
||||
aipodSpecDir?: string;
|
||||
auth?: ManagerAuthConfig;
|
||||
@@ -193,6 +212,7 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr
|
||||
const runnerDispatcherOptions = runnerDispatcherOptionsForRuntime(options.runnerDispatcherOptions);
|
||||
const kafkaOutboxRelayOptions = kafkaOutboxRelayOptionsForRuntime(options.kafkaOutboxRelayOptions);
|
||||
const resolvedRunnerJobDefaults = runnerJobDefaultsForRequest(runnerJobDefaults, sourceCommit);
|
||||
assertRunnerDispatcherRuntimeConfigured(runnerDispatcherOptions.enabled, runnerJobDefaults);
|
||||
const aipodSpecDir = options.aipodSpecDir ?? process.env.AGENTRUN_AIPOD_SPEC_DIR;
|
||||
const auth = options.auth ?? managerServerAuthConfigFromEnv();
|
||||
const authSummary = managerAuthSummary(auth);
|
||||
|
||||
+14
-11
@@ -4,6 +4,7 @@ import type { AgentRunStore } from "./store.js";
|
||||
import type { JsonRecord } from "../common/types.js";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactJson, redactText } from "../common/redaction.js";
|
||||
import { throwIfAborted, waitForChildProcess } from "./abortable-child.js";
|
||||
|
||||
export interface SessionPvcSpec {
|
||||
pvcName: string;
|
||||
@@ -24,7 +25,7 @@ export interface SessionPvcSummary extends JsonRecord {
|
||||
valuesPrinted: false;
|
||||
}
|
||||
|
||||
export type KubectlHandler = (input: { args: string[]; stdin?: string }) => { stdout: string; stderr: string; exitCode: number };
|
||||
export type KubectlHandler = (input: { args: string[]; stdin?: string; signal?: AbortSignal }) => { stdout: string; stderr: string; exitCode: number } | Promise<{ stdout: string; stderr: string; exitCode: number }>;
|
||||
|
||||
export interface SessionPvcOptions {
|
||||
storageClassName?: string;
|
||||
@@ -32,6 +33,7 @@ export interface SessionPvcOptions {
|
||||
kubectlCommand?: string;
|
||||
kubectlHandler?: KubectlHandler;
|
||||
defaultCodexRolloutSubdir?: string;
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
|
||||
const defaultStorageClassName = "local-path";
|
||||
@@ -81,10 +83,11 @@ export function buildSessionPvcSpec(input: { sessionId: string; namespace?: stri
|
||||
function resolveHandler(options: SessionPvcOptions): KubectlHandler {
|
||||
if (options.kubectlHandler) return options.kubectlHandler;
|
||||
const command = options.kubectlCommand ?? process.env.AGENTRUN_KUBECTL ?? "kubectl";
|
||||
return ({ args, stdin }) => spawnKubectl(command, args, stdin);
|
||||
return ({ args, stdin, signal }) => spawnKubectl(command, args, stdin, signal);
|
||||
}
|
||||
|
||||
function spawnKubectl(command: string, args: string[], stdinPayload?: string): { stdout: string; stderr: string; exitCode: number } {
|
||||
async function spawnKubectl(command: string, args: string[], stdinPayload?: string, signal?: AbortSignal): Promise<{ stdout: string; stderr: string; exitCode: number }> {
|
||||
throwIfAborted(signal, "kubectl");
|
||||
const child = spawn(command, args, { stdio: ["pipe", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
@@ -92,12 +95,12 @@ function spawnKubectl(command: string, args: string[], stdinPayload?: string): {
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
|
||||
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
|
||||
const completion = waitForChildProcess(child, { ...(signal ? { signal } : {}), label: "kubectl" });
|
||||
child.stdin.on("error", () => {});
|
||||
if (stdinPayload !== undefined) child.stdin.end(`${stdinPayload}\n`);
|
||||
else child.stdin.end();
|
||||
return new Promise<{ stdout: string; stderr: string; exitCode: number }>((resolve) => {
|
||||
child.on("error", () => resolve({ stdout, stderr, exitCode: 1 }));
|
||||
child.on("close", (code) => resolve({ stdout, stderr, exitCode: code ?? 1 }));
|
||||
}) as unknown as { stdout: string; stderr: string; exitCode: number };
|
||||
const result = await completion;
|
||||
return { stdout, stderr, exitCode: result.code ?? 1 };
|
||||
}
|
||||
|
||||
export async function createSessionPvc(input: { store: AgentRunStore; sessionId: string; namespace?: string; options: SessionPvcOptions }): Promise<SessionPvcSummary> {
|
||||
@@ -116,7 +119,7 @@ export async function createSessionPvc(input: { store: AgentRunStore; sessionId:
|
||||
spec: { accessModes: ["ReadWriteOnce"], storageClassName: spec.storageClassName, resources: { requests: { storage: spec.size } } },
|
||||
};
|
||||
const handler = resolveHandler(input.options);
|
||||
const result = await handler({ args: ["create", "-f", "-", "-o", "json"], stdin: JSON.stringify(manifest) });
|
||||
const result = await handler({ args: ["create", "-f", "-", "-o", "json"], stdin: JSON.stringify(manifest), ...(input.options.signal ? { signal: input.options.signal } : {}) });
|
||||
if (result.exitCode !== 0) {
|
||||
throw new AgentRunError("infra-failed", `kubectl create session PVC failed with code ${result.exitCode}`, { httpStatus: 502, details: redactJson({ stderr: redactText(result.stderr.slice(-4000)) }) });
|
||||
}
|
||||
@@ -155,7 +158,7 @@ export async function getSessionPvcSummary(input: { store: AgentRunStore; sessio
|
||||
const spec = buildSessionPvcSpec(input);
|
||||
const codexRolloutSubdir = session.codexRolloutSubdir ?? defaultSubdir;
|
||||
const handler = resolveHandler(input.options);
|
||||
const result = await handler({ args: ["get", "pvc", spec.pvcName, "-n", spec.namespace, "-o", "json"] });
|
||||
const result = await handler({ args: ["get", "pvc", spec.pvcName, "-n", spec.namespace, "-o", "json"], ...(input.options.signal ? { signal: input.options.signal } : {}) });
|
||||
if (result.exitCode !== 0) {
|
||||
const notFound = `${result.stdout}\n${result.stderr}`.toLowerCase().includes("notfound") || `${result.stdout}\n${result.stderr}`.toLowerCase().includes("not found");
|
||||
if (notFound) {
|
||||
@@ -195,7 +198,7 @@ export async function deleteSessionPvc(input: { store: AgentRunStore; sessionId:
|
||||
if (!session) throw new AgentRunError("schema-invalid", `session ${input.sessionId} was not found`, { httpStatus: 404 });
|
||||
const spec = buildSessionPvcSpec(input);
|
||||
const handler = resolveHandler(input.options);
|
||||
await handler({ args: ["delete", "pvc", spec.pvcName, "-n", spec.namespace, "--ignore-not-found"] });
|
||||
await handler({ args: ["delete", "pvc", spec.pvcName, "-n", spec.namespace, "--ignore-not-found"], ...(input.options.signal ? { signal: input.options.signal } : {}) });
|
||||
await input.store.markSessionStorageEvicted({ sessionId: input.sessionId, pvcName: spec.pvcName });
|
||||
return { pvcName: spec.pvcName, namespace: spec.namespace, storageKind: "evicted" };
|
||||
}
|
||||
@@ -249,7 +252,7 @@ export async function runSessionStorageGc(input: { store: AgentRunStore; options
|
||||
if (session.activeRunId || session.activeCommandId) { skipped++; continue; }
|
||||
if (session.storageKind !== "pvc" || !session.storagePvcName) { skipped++; continue; }
|
||||
try {
|
||||
await handler({ args: ["delete", "pvc", session.storagePvcName, "-n", session.storageNamespace ?? runtimeNamespace(), "--ignore-not-found"] });
|
||||
await handler({ args: ["delete", "pvc", session.storagePvcName, "-n", session.storageNamespace ?? runtimeNamespace(), "--ignore-not-found"], ...(input.options.signal ? { signal: input.options.signal } : {}) });
|
||||
await input.store.markSessionStorageEvicted({ sessionId: session.sessionId, pvcName: session.storagePvcName });
|
||||
deleted++;
|
||||
deletedPvcNames.push(session.storagePvcName);
|
||||
|
||||
+66
-27
@@ -1,4 +1,4 @@
|
||||
import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, FailureKind, JsonRecord, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, QueueTaskSummary, RunEvent, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, SessionEventPage, SessionListResult, SessionListState, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
|
||||
import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, FailureKind, JsonRecord, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, QueueTaskSummary, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, SessionEventPage, SessionListResult, SessionListState, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { newId, nowIso, stableHash } from "../common/validation.js";
|
||||
import { redactJson } from "../common/redaction.js";
|
||||
@@ -26,6 +26,12 @@ export interface EventOutboxConfig {
|
||||
source: string | null;
|
||||
}
|
||||
|
||||
export interface DurableQueueClaim {
|
||||
id: string;
|
||||
leaseOwner: string | null;
|
||||
attemptCount: number;
|
||||
}
|
||||
|
||||
export interface AgentRunStore {
|
||||
health(): MaybePromise<StoreHealth>;
|
||||
createRun(input: CreateRunInput): MaybePromise<RunRecord>;
|
||||
@@ -37,13 +43,13 @@ export interface AgentRunStore {
|
||||
listCommands(runId: string, afterSeq: number, limit: number): MaybePromise<CommandRecord[]>;
|
||||
getRunnerDispatchIntent(commandId: string): MaybePromise<RunnerDispatchIntentRecord | null>;
|
||||
claimRunnerDispatchIntents(input: { owner: string; leaseMs: number; limit: number }): MaybePromise<RunnerDispatchIntentRecord[]>;
|
||||
completeRunnerDispatchIntent(intentId: string, runnerJobId: string): MaybePromise<RunnerDispatchIntentRecord>;
|
||||
retryRunnerDispatchIntent(intentId: string, error: JsonRecord, nextAttemptAt: string): MaybePromise<RunnerDispatchIntentRecord>;
|
||||
terminalizeRunnerDispatchIntent(intentId: string, error: JsonRecord): MaybePromise<JsonRecord>;
|
||||
completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): MaybePromise<RunnerDispatchIntentRecord>;
|
||||
retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): MaybePromise<RunnerDispatchIntentRecord>;
|
||||
terminalizeRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord): MaybePromise<JsonRecord>;
|
||||
runnerDispatchStatus(): MaybePromise<JsonRecord>;
|
||||
claimKafkaEventOutbox(input: { owner: string; leaseMs: number; limit: number }): MaybePromise<KafkaEventOutboxRecord[]>;
|
||||
completeKafkaEventOutbox(outboxId: string): MaybePromise<KafkaEventOutboxRecord>;
|
||||
retryKafkaEventOutbox(outboxId: string, error: JsonRecord, nextAttemptAt: string): MaybePromise<KafkaEventOutboxRecord>;
|
||||
completeKafkaEventOutbox(claim: DurableQueueClaim): MaybePromise<KafkaEventOutboxRecord>;
|
||||
retryKafkaEventOutbox(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string): MaybePromise<KafkaEventOutboxRecord>;
|
||||
kafkaEventOutboxStatus(): MaybePromise<JsonRecord>;
|
||||
registerRunner(input: Partial<RunnerRecord>): MaybePromise<RunnerRecord>;
|
||||
listRunnerJobs(runId: string, commandId?: string): MaybePromise<RunnerJobRecord[]>;
|
||||
@@ -258,23 +264,30 @@ export class MemoryAgentRunStore implements AgentRunStore {
|
||||
});
|
||||
}
|
||||
|
||||
completeRunnerDispatchIntent(intentId: string, runnerJobId: string): RunnerDispatchIntentRecord {
|
||||
const intent = this.requireRunnerDispatchIntent(intentId);
|
||||
completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): RunnerDispatchIntentRecord {
|
||||
const intent = this.requireRunnerDispatchIntent(claim.id);
|
||||
assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching");
|
||||
const event = this.prepareEvent(intent.runId, "backend_status", eventPayload);
|
||||
const at = nowIso();
|
||||
const updated: RunnerDispatchIntentRecord = { ...intent, state: "dispatched", runnerJobId, leaseOwner: null, leaseExpiresAt: null, lastError: null, dispatchedAt: at, updatedAt: at };
|
||||
this.runnerDispatchIntents.set(intentId, updated);
|
||||
const updated: RunnerDispatchIntentRecord = { ...intent, state: "dispatched", ...completion, leaseOwner: null, leaseExpiresAt: null, lastError: null, dispatchedAt: at, updatedAt: at };
|
||||
this.runnerDispatchIntents.set(intent.id, updated);
|
||||
this.commitPreparedEvent(event);
|
||||
return updated;
|
||||
}
|
||||
|
||||
retryRunnerDispatchIntent(intentId: string, error: JsonRecord, nextAttemptAt: string): RunnerDispatchIntentRecord {
|
||||
const intent = this.requireRunnerDispatchIntent(intentId);
|
||||
retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): RunnerDispatchIntentRecord {
|
||||
const intent = this.requireRunnerDispatchIntent(claim.id);
|
||||
assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching");
|
||||
const event = this.prepareEvent(intent.runId, "backend_status", eventPayload);
|
||||
const updated: RunnerDispatchIntentRecord = { ...intent, state: "retry", leaseOwner: null, leaseExpiresAt: null, nextAttemptAt, lastError: redactJson(error), updatedAt: nowIso() };
|
||||
this.runnerDispatchIntents.set(intentId, updated);
|
||||
this.runnerDispatchIntents.set(intent.id, updated);
|
||||
this.commitPreparedEvent(event);
|
||||
return updated;
|
||||
}
|
||||
|
||||
terminalizeRunnerDispatchIntent(intentId: string, error: JsonRecord): JsonRecord {
|
||||
const intent = this.requireRunnerDispatchIntent(intentId);
|
||||
terminalizeRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord): JsonRecord {
|
||||
const intent = this.requireRunnerDispatchIntent(claim.id);
|
||||
assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching");
|
||||
const at = nowIso();
|
||||
const failure = redactJson(error);
|
||||
const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed";
|
||||
@@ -313,18 +326,20 @@ export class MemoryAgentRunStore implements AgentRunStore {
|
||||
});
|
||||
}
|
||||
|
||||
completeKafkaEventOutbox(outboxId: string): KafkaEventOutboxRecord {
|
||||
const item = this.requireKafkaEventOutbox(outboxId);
|
||||
completeKafkaEventOutbox(claim: DurableQueueClaim): KafkaEventOutboxRecord {
|
||||
const item = this.requireKafkaEventOutbox(claim.id);
|
||||
assertMemoryClaim(item, claim, "Kafka event outbox", !item.deliveredAt);
|
||||
const at = nowIso();
|
||||
const updated: KafkaEventOutboxRecord = { ...item, deliveredAt: at, leaseOwner: null, leaseExpiresAt: null, lastError: null, updatedAt: at };
|
||||
this.kafkaEventOutbox.set(outboxId, updated);
|
||||
this.kafkaEventOutbox.set(item.id, updated);
|
||||
return updated;
|
||||
}
|
||||
|
||||
retryKafkaEventOutbox(outboxId: string, error: JsonRecord, nextAttemptAt: string): KafkaEventOutboxRecord {
|
||||
const item = this.requireKafkaEventOutbox(outboxId);
|
||||
retryKafkaEventOutbox(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string): KafkaEventOutboxRecord {
|
||||
const item = this.requireKafkaEventOutbox(claim.id);
|
||||
assertMemoryClaim(item, claim, "Kafka event outbox", !item.deliveredAt);
|
||||
const updated: KafkaEventOutboxRecord = { ...item, leaseOwner: null, leaseExpiresAt: null, nextAttemptAt, lastError: redactJson(error), updatedAt: nowIso() };
|
||||
this.kafkaEventOutbox.set(outboxId, updated);
|
||||
this.kafkaEventOutbox.set(item.id, updated);
|
||||
return updated;
|
||||
}
|
||||
|
||||
@@ -436,22 +451,39 @@ export class MemoryAgentRunStore implements AgentRunStore {
|
||||
}
|
||||
|
||||
appendEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): RunEvent {
|
||||
const prepared = this.prepareEvent(runId, type, payload);
|
||||
this.commitPreparedEvent(prepared);
|
||||
return prepared.event;
|
||||
}
|
||||
|
||||
protected beforeEventCommit(_event: RunEvent): void {}
|
||||
|
||||
private prepareEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): { run: RunRecord; event: RunEvent; outbox: KafkaEventOutboxRecord | null } {
|
||||
const run = this.getRun(runId);
|
||||
const eventType = requireEventType(type);
|
||||
const fenced = fenceLateEventForCancelledRun(run, eventType, payload);
|
||||
const eventPayload = normalizeRunEventPayload(fenced.type, fenced.payload);
|
||||
const events = this.eventsByRun.get(runId) ?? [];
|
||||
const event: RunEvent = { id: newId("evt"), runId, seq: events.length + 1, type: fenced.type, payload: redactJson(eventPayload), createdAt: nowIso() };
|
||||
events.push(event);
|
||||
this.eventsByRun.set(runId, events);
|
||||
let outbox: KafkaEventOutboxRecord | null = null;
|
||||
if (this.eventOutboxConfig.enabled) {
|
||||
const command = commandForEvent(this.commands, event);
|
||||
this.kafkaOutboxSeq += 1;
|
||||
const outbox = buildKafkaEventOutboxRecord({ source: this.eventOutboxConfig.source as string, topic: this.eventOutboxConfig.topic as string, outboxSeq: this.kafkaOutboxSeq, run, command, event });
|
||||
outbox = buildKafkaEventOutboxRecord({ source: this.eventOutboxConfig.source as string, topic: this.eventOutboxConfig.topic as string, outboxSeq: this.kafkaOutboxSeq + 1, run, command, event });
|
||||
}
|
||||
this.beforeEventCommit(event);
|
||||
return { run, event, outbox };
|
||||
}
|
||||
|
||||
private commitPreparedEvent(prepared: { run: RunRecord; event: RunEvent; outbox: KafkaEventOutboxRecord | null }): void {
|
||||
const { run, event, outbox } = prepared;
|
||||
const events = this.eventsByRun.get(run.id) ?? [];
|
||||
events.push(event);
|
||||
this.eventsByRun.set(run.id, events);
|
||||
if (outbox) {
|
||||
this.kafkaOutboxSeq = outbox.outboxSeq;
|
||||
this.kafkaEventOutbox.set(outbox.id, outbox);
|
||||
}
|
||||
this.touchSessionForRun(this.getRun(runId), { lastEventSeq: event.seq, lastActivityAt: event.createdAt }, { bumpVersion: false, at: event.createdAt });
|
||||
return event;
|
||||
this.touchSessionForRun(this.getRun(run.id), { lastEventSeq: event.seq, lastActivityAt: event.createdAt }, { bumpVersion: false, at: event.createdAt });
|
||||
}
|
||||
|
||||
finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): RunRecord {
|
||||
@@ -1154,6 +1186,13 @@ function commandForEvent(commands: Map<string, CommandRecord>, event: RunEvent):
|
||||
return null;
|
||||
}
|
||||
|
||||
function assertMemoryClaim(record: DurableQueueClaim & { leaseExpiresAt?: string | null }, claim: DurableQueueClaim, component: string, stateMatches: boolean): void {
|
||||
const leaseExpiresAt = Date.parse(record.leaseExpiresAt ?? "");
|
||||
if (!stateMatches || !claim.leaseOwner || record.leaseOwner !== claim.leaseOwner || record.attemptCount !== claim.attemptCount || !Number.isFinite(leaseExpiresAt) || leaseExpiresAt <= Date.now()) {
|
||||
throw new AgentRunError("runner-lease-conflict", `${component} claim is stale`, { httpStatus: 409, details: { id: claim.id, attemptCount: claim.attemptCount, valuesPrinted: false } });
|
||||
}
|
||||
}
|
||||
|
||||
function workQueueStatus<T extends JsonRecord>(items: T[], component: string, completed: (item: T) => boolean = (item) => ["dispatched", "failed", "cancelled"].includes(String(item.state ?? ""))): JsonRecord {
|
||||
const pending = items.filter((item) => !completed(item));
|
||||
const retrying = pending.filter((item) => Number(item.attemptCount ?? 0) > 0);
|
||||
|
||||
@@ -13,17 +13,19 @@ const selfTest: SelfTestCase = async () => {
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "infra-failed" && error.message.includes("DATABASE_URL is required"),
|
||||
);
|
||||
const postgresContract = postgresMigrationContract();
|
||||
assert.equal(postgresContract.latestMigrationId, "012_v02_durable_dispatch_kafka_outbox");
|
||||
assert.equal(postgresContract.latestMigrationId, "013_v02_runner_dispatch_outcome");
|
||||
assert.equal((postgresContract.migrationIds as string[]).includes("008_v01_dsflash_go_backend_profile"), true);
|
||||
assert.equal((postgresContract.migrationIds as string[]).includes("009_v01_dsflash_go_model_catalog"), true);
|
||||
assert.equal((postgresContract.migrationIds as string[]).includes("010_v01_queue_session_ref"), true);
|
||||
assert.equal((postgresContract.migrationIds as string[]).includes("011_v01_cancel_lifecycle"), true);
|
||||
assert.equal((postgresContract.migrationIds as string[]).includes("012_v02_durable_dispatch_kafka_outbox"), true);
|
||||
assert.equal((postgresContract.migrationIds as string[]).includes("013_v02_runner_dispatch_outcome"), true);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record<string, string>)["008_v01_dsflash_go_backend_profile"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["011_v01_cancel_lifecycle"] === "string" && (postgresContract.checksums as Record<string, string>)["011_v01_cancel_lifecycle"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["012_v02_durable_dispatch_kafka_outbox"] === "string" && (postgresContract.checksums as Record<string, string>)["012_v02_durable_dispatch_kafka_outbox"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["013_v02_runner_dispatch_outcome"] === "string" && (postgresContract.checksums as Record<string, string>)["013_v02_runner_dispatch_outcome"].length > 0);
|
||||
assert.equal((postgresContract.checksums as Record<string, string>)["002_v01_backend_profiles"], "928b5c490cc4539cb64ecef34784557601b2724fa2870570f16a53576804e49c");
|
||||
assert.ok(Array.isArray(postgresContract.requiredTables));
|
||||
assert.ok(postgresContract.requiredTables.includes("agentrun_schema_migrations"));
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { chmod } from "node:fs/promises";
|
||||
import { access, chmod, rm } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import type { CreateRunInput } from "../../common/types.js";
|
||||
import type { CreateRunInput, RunEvent } from "../../common/types.js";
|
||||
import { validateCreateCommand } from "../../common/validation.js";
|
||||
import { dispatchRunnerIntentsOnce, type RunnerDispatcherOptions } from "../../mgr/runner-dispatcher.js";
|
||||
import { MemoryAgentRunStore } from "../../mgr/store.js";
|
||||
@@ -27,7 +27,7 @@ const selfTest: SelfTestCase = async (context) => {
|
||||
type: "turn",
|
||||
idempotencyKey: "durable-adopt",
|
||||
payload: { prompt: "adopt orphan Kubernetes Job" },
|
||||
dispatch: { kind: "kubernetes-job", input: { image } },
|
||||
dispatch: { kind: "kubernetes-job", input: { image, namespace: "agentrun-v02", serviceAccountName: "agentrun-v02-runner", transientEnv: hwlabRuntimeEnv() } },
|
||||
}));
|
||||
const result = await dispatchRunnerIntentsOnce({ store, defaults: defaults(fakeKubectl), options: dispatcherOptions(3) });
|
||||
assert.equal(result.dispatchedCount, 1, JSON.stringify({ result, intent: store.getRunnerDispatchIntent(command.id) }));
|
||||
@@ -42,6 +42,11 @@ const selfTest: SelfTestCase = async (context) => {
|
||||
assert.equal(jobs[0]?.result.mutation, false);
|
||||
assert.equal((jobs[0]?.result.kubernetes as { created?: unknown }).created, false);
|
||||
await assertHttpAdmissionContract(fakeKubectl);
|
||||
await assertActiveRunnerReuseHasNoDanglingJob(fakeKubectl);
|
||||
await assertHungKubectlTimesOutAndRecovers(fakeKubectl);
|
||||
await assertClaimsAreLeasedJustInTime(fakeKubectl);
|
||||
await assertSettleEventFailureDoesNotHalfCommit(fakeKubectl);
|
||||
await assertEnabledDispatcherRejectsLegacyDefaults();
|
||||
|
||||
process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "fail";
|
||||
const failedRun = store.createRun({ ...runInput(), sessionRef: null });
|
||||
@@ -49,7 +54,7 @@ const selfTest: SelfTestCase = async (context) => {
|
||||
type: "turn",
|
||||
idempotencyKey: "durable-terminal",
|
||||
payload: { prompt: "terminalize permanent dispatch failure" },
|
||||
dispatch: { kind: "kubernetes-job", input: { image } },
|
||||
dispatch: { kind: "kubernetes-job", input: { image, namespace: "agentrun-v02", serviceAccountName: "agentrun-v02-runner", transientEnv: hwlabRuntimeEnv() } },
|
||||
}));
|
||||
const failed = await dispatchRunnerIntentsOnce({ store, defaults: defaults(fakeKubectl), options: dispatcherOptions(1) });
|
||||
assert.equal(failed.failedCount, 1);
|
||||
@@ -60,7 +65,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", "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", "permanent-failure-terminal"] };
|
||||
};
|
||||
|
||||
async function assertHttpAdmissionContract(kubectlCommand: string): Promise<void> {
|
||||
@@ -81,7 +86,7 @@ async function assertHttpAdmissionContract(kubectlCommand: string): Promise<void
|
||||
type: "turn",
|
||||
idempotencyKey: "http-durable-admission",
|
||||
payload: { prompt: "durable HTTP admission" },
|
||||
dispatch: { kind: "kubernetes-job", input: { image } },
|
||||
dispatch: { kind: "kubernetes-job", input: { image, namespace: "agentrun-v02", serviceAccountName: "agentrun-v02-runner", transientEnv: hwlabRuntimeEnv() } },
|
||||
}) as { dispatchIntent?: { id?: string; state?: string; runnerJobId?: string; attemptCount?: number; durable?: boolean }; dispatch?: unknown };
|
||||
assert.match(command.dispatchIntent?.id ?? "", /^dispatch_[a-f0-9]{24}$/u);
|
||||
assert.equal(command.dispatchIntent?.state, "pending");
|
||||
@@ -108,7 +113,115 @@ function defaults(kubectlCommand: string): RunnerJobDefaults {
|
||||
}
|
||||
|
||||
function dispatcherOptions(maxAttempts: number): RunnerDispatcherOptions {
|
||||
return { enabled: true, intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, owner: "selftest-dispatcher" };
|
||||
return { enabled: true, intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "selftest-dispatcher" };
|
||||
}
|
||||
|
||||
async function assertActiveRunnerReuseHasNoDanglingJob(kubectlCommand: string): Promise<void> {
|
||||
const store = new MemoryAgentRunStore();
|
||||
const run = store.createRun(runInput());
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
store.claimRun(run.id, "runner_existing", 60_000);
|
||||
const result = await dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options: dispatcherOptions(3) });
|
||||
assert.equal(result.dispatchedCount, 1);
|
||||
const intent = store.getRunnerDispatchIntent(command.id);
|
||||
assert.equal(intent?.dispatchOutcome, "active-runner-reused");
|
||||
assert.equal(intent?.actualRunnerJobId, null);
|
||||
assert.equal(intent?.activeRunnerId, "runner_existing");
|
||||
assert.match(intent?.runnerJobId ?? "", /^rjob_[a-f0-9]{24}$/u);
|
||||
assert.equal(store.listRunnerJobs(run.id, command.id).length, 0);
|
||||
}
|
||||
|
||||
async function assertHungKubectlTimesOutAndRecovers(kubectlCommand: string): Promise<void> {
|
||||
const store = new MemoryAgentRunStore();
|
||||
const run = store.createRun(runInput());
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "hang";
|
||||
const marker = `${process.env.AGENTRUN_SELFTEST_KUBECTL_STATE}.hang-started`;
|
||||
await rm(marker, { force: true });
|
||||
const options = { ...dispatcherOptions(3), leaseMs: 2_000, retryBackoffMs: 10, attemptTimeoutMs: 250 };
|
||||
const firstCycle = dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options });
|
||||
await waitForFile(marker, 1_000);
|
||||
process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "adopt";
|
||||
const firstResult = await firstCycle;
|
||||
assert.equal(firstResult.retryCount, 1, JSON.stringify({ firstResult, intent: store.getRunnerDispatchIntent(command.id) }));
|
||||
assert.equal(store.getRunnerDispatchIntent(command.id)?.state, "retry");
|
||||
await new Promise((resolve) => setTimeout(resolve, 15));
|
||||
assert.equal((await dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options })).dispatchedCount, 1);
|
||||
assert.equal(store.getRunnerDispatchIntent(command.id)?.attemptCount, 2);
|
||||
assert.equal(store.getRunnerDispatchIntent(command.id)?.dispatchOutcome, "kubernetes-job");
|
||||
}
|
||||
|
||||
async function assertClaimsAreLeasedJustInTime(kubectlCommand: string): Promise<void> {
|
||||
const store = new MemoryAgentRunStore();
|
||||
const firstRun = store.createRun(runInput());
|
||||
const first = store.createCommand(firstRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
const secondRun = store.createRun(runInput());
|
||||
const second = store.createCommand(secondRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "hang";
|
||||
const marker = `${process.env.AGENTRUN_SELFTEST_KUBECTL_STATE}.hang-started`;
|
||||
await rm(marker, { force: true });
|
||||
const options = { ...dispatcherOptions(3), batchSize: 2, leaseMs: 2_000, retryBackoffMs: 1_000, attemptTimeoutMs: 250 };
|
||||
const cycle = dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options });
|
||||
await waitForFile(marker, 1_000);
|
||||
assert.equal(store.getRunnerDispatchIntent(second.id)?.state, "pending");
|
||||
assert.equal(store.getRunnerDispatchIntent(second.id)?.attemptCount, 0);
|
||||
process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "fail";
|
||||
const result = await cycle;
|
||||
assert.equal(result.claimedCount, 2);
|
||||
assert.equal(store.getRunnerDispatchIntent(second.id)?.attemptCount, 1);
|
||||
assert.equal(store.getRunnerDispatchIntent(second.id)?.state, "retry");
|
||||
}
|
||||
|
||||
async function assertSettleEventFailureDoesNotHalfCommit(kubectlCommand: string): Promise<void> {
|
||||
const store = new FailingDispatchEventStore({ eventOutbox: { enabled: true, topic: "agentrun.event.v1", source: "selftest" } });
|
||||
const run = store.createRun(runInput());
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "adopt";
|
||||
await assert.rejects(() => dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options: dispatcherOptions(3) }), /simulated dispatch event write failure/u);
|
||||
assert.equal(store.getRunnerDispatchIntent(command.id)?.state, "dispatching");
|
||||
assert.equal(store.listEvents(run.id, 0, 100).some((event) => event.payload.phase === "runner-dispatch-completed" || event.payload.phase === "runner-dispatch-retry"), false);
|
||||
assert.equal(store.kafkaEventOutboxStatus().backlogCount, store.listEvents(run.id, 0, 100).length);
|
||||
}
|
||||
|
||||
class FailingDispatchEventStore extends MemoryAgentRunStore {
|
||||
protected override beforeEventCommit(event: RunEvent): void {
|
||||
if (event.payload.phase === "runner-dispatch-completed" || event.payload.phase === "runner-dispatch-retry") throw new Error("simulated dispatch event write failure");
|
||||
}
|
||||
}
|
||||
|
||||
async function assertEnabledDispatcherRejectsLegacyDefaults(): Promise<void> {
|
||||
const keys = ["AGENTRUN_RUNTIME_NAMESPACE", "AGENTRUN_INTERNAL_MGR_URL", "AGENTRUN_RUNNER_IMAGE", "AGENTRUN_RUNNER_SERVICE_ACCOUNT", "AGENTRUN_RUNNER_JOB_NAME_PREFIX", "AGENTRUN_LANE"] as const;
|
||||
const previous = Object.fromEntries(keys.map((key) => [key, process.env[key]]));
|
||||
keys.forEach((key) => { delete process.env[key]; });
|
||||
try {
|
||||
await assert.rejects(
|
||||
() => startManagerServer({ store: new MemoryAgentRunStore(), sourceCommit: "selftest", runnerDispatcherOptions: dispatcherOptions(3), kafkaOutboxRelayOptions: { enabled: false } }),
|
||||
(error) => error instanceof Error && error.message.includes("runtime configuration is incomplete"),
|
||||
);
|
||||
} finally {
|
||||
keys.forEach((key) => restoreEnv(key, previous[key]));
|
||||
}
|
||||
await assert.rejects(
|
||||
() => startManagerServer({ store: new MemoryAgentRunStore(), sourceCommit: "selftest", runnerJobDefaults: defaults("kubectl"), runnerDispatcherOptions: { ...dispatcherOptions(3), leaseMs: 1_000, attemptTimeoutMs: 1_000 }, kafkaOutboxRelayOptions: { enabled: false } }),
|
||||
(error) => error instanceof Error && error.message.includes("strictly less"),
|
||||
);
|
||||
}
|
||||
|
||||
async function waitFor(predicate: () => boolean, timeoutMs: number, message: string | (() => string)): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (predicate()) return;
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
throw new Error(typeof message === "string" ? message : message());
|
||||
}
|
||||
|
||||
async function waitForFile(file: string, timeoutMs: number): Promise<void> {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
try { await access(file); return; } catch { await new Promise((resolve) => setTimeout(resolve, 10)); }
|
||||
}
|
||||
throw new Error(`timed out waiting for ${path.basename(file)}`);
|
||||
}
|
||||
|
||||
function runInput(): CreateRunInput {
|
||||
@@ -125,6 +238,21 @@ function runInput(): CreateRunInput {
|
||||
};
|
||||
}
|
||||
|
||||
function hwlabRuntimeEnv(): Array<{ name: string; value: string; sensitive: false }> {
|
||||
return [
|
||||
{ name: "HWLAB_RUNTIME_API_URL", value: "http://runtime-api.test", sensitive: false },
|
||||
{ name: "HWLAB_RUNTIME_WEB_URL", value: "http://runtime-web.test", sensitive: false },
|
||||
{ name: "HWLAB_RUNTIME_NAMESPACE", value: "hwlab-v03", sensitive: false },
|
||||
{ name: "HWLAB_RUNTIME_LANE", value: "v03", sensitive: false },
|
||||
{ name: "HWLAB_RUNTIME_ENDPOINT_SOURCE", value: "runtime-namespace", sensitive: false },
|
||||
{ name: "HWLAB_RUNTIME_ENDPOINT_LOCKED", value: "1", sensitive: false },
|
||||
{ name: "HWLAB_CODE_AGENT_ASSEMBLED_RUNTIME", value: "1", sensitive: false },
|
||||
{ name: "UNIDESK_MAIN_SERVER_IP", value: "https://unidesk.example.test", sensitive: false },
|
||||
{ name: "HWLAB_CODE_AGENT_PROVIDER_PROFILE", value: "codex", sensitive: false },
|
||||
{ name: "HWLAB_CODE_AGENT_PARENT_TRACE_ID", value: "trc_selftest", sensitive: false },
|
||||
];
|
||||
}
|
||||
|
||||
function restoreEnv(key: string, value: string | undefined): void {
|
||||
if (value === undefined) delete process.env[key];
|
||||
else process.env[key] = value;
|
||||
|
||||
@@ -46,7 +46,7 @@ const selfTest: SelfTestCase = async () => {
|
||||
payload: { prompt: "exercise durable dispatch", traceId: "trc_durable" },
|
||||
dispatch: {
|
||||
kind: "kubernetes-job",
|
||||
input: { attemptId: "attempt_durable", runnerId: "runner_durable", imageRef: { kind: "env-image-dockerfile", repoUrl: "git@example.test/repo.git", commitId: "abc", dockerfilePath: "Containerfile" } },
|
||||
input: { attemptId: "attempt_durable", runnerId: "runner_durable", imageRef: { kind: "env-image-dockerfile", repoUrl: "git@example.test:repo.git", commitId: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", dockerfilePath: "Containerfile" } },
|
||||
},
|
||||
});
|
||||
const command = store.createCommand(run.id, commandInput);
|
||||
@@ -58,13 +58,39 @@ const selfTest: SelfTestCase = async () => {
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { transientEnv: [{ name: "HWLAB_API_KEY", value: "secret", sensitive: true }] } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied",
|
||||
);
|
||||
assert.throws(
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { ownerApiKey: "secret" } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied",
|
||||
);
|
||||
assert.throws(
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { transientEnv: [{ name: "HWLAB_API_KEY", value: "secret", sensitive: false }] } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied",
|
||||
);
|
||||
assert.throws(
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { imageRef: { metadata: { accessToken: "secret" } } } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied",
|
||||
);
|
||||
assert.throws(
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { managerUrl: "https://owner:secret@manager.example.test" } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied",
|
||||
);
|
||||
assert.throws(
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { imageRef: { kind: "env-image-dockerfile", repoUrl: "https://owner:secret@example.test/repo.git", commitId: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", dockerfilePath: "Containerfile" } } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "schema-invalid",
|
||||
);
|
||||
assert.throws(
|
||||
() => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { imageRef: { kind: "env-image-dockerfile", repoUrl: "git@example.test:repo.git", commitId: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", dockerfilePath: "Containerfile", extra: true } } } }),
|
||||
(error) => error instanceof AgentRunError && error.failureKind === "schema-invalid",
|
||||
);
|
||||
const normalizedDispatch = validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { namespace: "agentrun-v02", transientEnv: [{ name: "HWLAB_RUNTIME_LANE", value: "v02", sensitive: false }] } } }).dispatch?.input;
|
||||
assert.deepEqual(normalizedDispatch, { namespace: "agentrun-v02", transientEnv: [{ name: "HWLAB_RUNTIME_LANE", value: "v02", sensitive: false }] });
|
||||
|
||||
const firstClaim = store.claimKafkaEventOutbox({ owner: "first", leaseMs: 60_000, limit: 100 });
|
||||
assert.equal(firstClaim.length, 1);
|
||||
assert.equal(firstClaim[0]?.outboxSeq, 1);
|
||||
store.retryKafkaEventOutbox(firstClaim[0]?.id ?? "", { code: "broker-unavailable" }, new Date(Date.now() + 60_000).toISOString());
|
||||
store.retryKafkaEventOutbox(firstClaim[0]!, { code: "broker-unavailable" }, new Date(Date.now() + 20).toISOString());
|
||||
assert.deepEqual(store.claimKafkaEventOutbox({ owner: "second", leaseMs: 60_000, limit: 100 }), []);
|
||||
store.retryKafkaEventOutbox(firstClaim[0]?.id ?? "", { code: "retry-now" }, new Date(0).toISOString());
|
||||
await delay(25);
|
||||
|
||||
const published: JsonRecord[] = [];
|
||||
const publish = async (_config: AgentRunKafkaConfig, message: { value: JsonRecord }): Promise<void> => { published.push(message.value); };
|
||||
@@ -80,7 +106,7 @@ const selfTest: SelfTestCase = async () => {
|
||||
|
||||
const claimedIntent = store.claimRunnerDispatchIntents({ owner: "terminalizer", leaseMs: 60_000, limit: 1 })[0];
|
||||
assert.ok(claimedIntent);
|
||||
const terminalized = store.terminalizeRunnerDispatchIntent(claimedIntent.id, { message: "kubectl permanently unavailable", failureKind: "infra-failed", valuesPrinted: false });
|
||||
const terminalized = store.terminalizeRunnerDispatchIntent(claimedIntent, { message: "kubectl permanently unavailable", failureKind: "infra-failed", valuesPrinted: false });
|
||||
assert.equal((terminalized.intent as JsonRecord).state, "failed");
|
||||
assert.equal(store.getCommand(command.id).state, "failed");
|
||||
assert.equal(store.getRun(run.id).status, "failed");
|
||||
@@ -88,10 +114,39 @@ const selfTest: SelfTestCase = async () => {
|
||||
assert.deepEqual(store.listEvents(run.id, 0, 100).slice(-3).map((event) => event.payload.phase ?? event.type), ["runner-dispatch-failed", "command-terminal", "terminal_status"]);
|
||||
assert.throws(() => store.createCommand(run.id, commandInput), (error) => error instanceof AgentRunError && error.httpStatus !== 200);
|
||||
|
||||
await assertStaleClaimsCannotOverwrite();
|
||||
await assertProducerRecreatedAfterDisconnected();
|
||||
return { name: "kafka-durable-outbox", tests: ["explicit-enable", "canonical-event", "durable-head-of-line", "atomic-dispatch-terminal", "producer-reconnect"] };
|
||||
return { name: "kafka-durable-outbox", tests: ["explicit-enable", "canonical-event", "durable-head-of-line", "atomic-dispatch-terminal", "stale-claim-fencing", "producer-reconnect"] };
|
||||
};
|
||||
|
||||
async function assertStaleClaimsCannotOverwrite(): Promise<void> {
|
||||
const store = new MemoryAgentRunStore({ eventOutbox: { enabled: true, topic: config.agentrunEventTopic, source: config.clientId } });
|
||||
const run = store.createRun(runInput("ses_stale_claim"));
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } }));
|
||||
const dispatchA = store.claimRunnerDispatchIntents({ owner: "worker-a", leaseMs: 1, limit: 1 })[0];
|
||||
assert.ok(dispatchA);
|
||||
await delay(5);
|
||||
const dispatchB = store.claimRunnerDispatchIntents({ owner: "worker-b", leaseMs: 60_000, limit: 1 })[0];
|
||||
assert.ok(dispatchB);
|
||||
const completion = { dispatchOutcome: "kubernetes-job" as const, actualRunnerJobId: dispatchB.runnerJobId, activeRunnerId: null };
|
||||
store.completeRunnerDispatchIntent(dispatchB, completion, { phase: "runner-dispatch-completed" });
|
||||
assert.throws(() => store.completeRunnerDispatchIntent(dispatchA, completion, { phase: "runner-dispatch-completed" }), staleClaim);
|
||||
assert.throws(() => store.retryRunnerDispatchIntent(dispatchA, {}, new Date().toISOString(), { phase: "runner-dispatch-retry" }), staleClaim);
|
||||
assert.throws(() => store.terminalizeRunnerDispatchIntent(dispatchA, {}), staleClaim);
|
||||
assert.equal(store.getRunnerDispatchIntent(command.id)?.state, "dispatched");
|
||||
|
||||
const outboxStore = new MemoryAgentRunStore({ eventOutbox: { enabled: true, topic: config.agentrunEventTopic, source: config.clientId } });
|
||||
outboxStore.createRun(runInput("ses_stale_outbox"));
|
||||
const outboxA = outboxStore.claimKafkaEventOutbox({ owner: "relay-a", leaseMs: 1, limit: 1 })[0];
|
||||
assert.ok(outboxA);
|
||||
await delay(5);
|
||||
const outboxB = outboxStore.claimKafkaEventOutbox({ owner: "relay-b", leaseMs: 60_000, limit: 1 })[0];
|
||||
assert.ok(outboxB);
|
||||
outboxStore.completeKafkaEventOutbox(outboxB);
|
||||
assert.throws(() => outboxStore.retryKafkaEventOutbox(outboxA, {}, new Date().toISOString()), staleClaim);
|
||||
assert.equal(outboxStore.kafkaEventOutboxStatus().backlogCount, 0);
|
||||
}
|
||||
|
||||
async function assertProducerRecreatedAfterDisconnected(): Promise<void> {
|
||||
let factoryCalls = 0;
|
||||
let disconnectCalls = 0;
|
||||
@@ -154,4 +209,12 @@ function runInput(sessionId: string): CreateRunInput {
|
||||
};
|
||||
}
|
||||
|
||||
function staleClaim(error: unknown): boolean {
|
||||
return error instanceof AgentRunError && error.failureKind === "runner-lease-conflict";
|
||||
}
|
||||
|
||||
function delay(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
|
||||
export default selfTest;
|
||||
|
||||
@@ -11,6 +11,12 @@ if (!statePath) {
|
||||
}
|
||||
|
||||
if (args[0] === "create") {
|
||||
if (mode === "hang") {
|
||||
await writeFile(`${statePath}.hang-started`, String(process.pid));
|
||||
process.on("SIGTERM", () => {});
|
||||
setInterval(() => {}, 1_000);
|
||||
await new Promise(() => {});
|
||||
}
|
||||
if (mode === "fail") {
|
||||
console.error("simulated Kubernetes API outage");
|
||||
process.exit(1);
|
||||
|
||||
Reference in New Issue
Block a user