diff --git a/src/common/aipod-image-ref-validation.ts b/src/common/aipod-image-ref-validation.ts new file mode 100644 index 0000000..f9ae7a5 --- /dev/null +++ b/src/common/aipod-image-ref-validation.ts @@ -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 }); +} diff --git a/src/common/env-image-ref.ts b/src/common/env-image-ref.ts index 9852cd0..b82c3bd 100644 --- a/src/common/env-image-ref.ts +++ b/src/common/env-image-ref.ts @@ -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; } diff --git a/src/common/types.ts b/src/common/types.ts index 68dd290..fe1ad40 100644 --- a/src/common/types.ts +++ b/src/common/types.ts @@ -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; } diff --git a/src/common/validation.ts b/src/common/validation.ts index ff1ae3b..e411a6c 100644 --- a/src/common/validation.ts +++ b/src/common/validation.ts @@ -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(); + 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 { diff --git a/src/mgr/abortable-child.ts b/src/mgr/abortable-child.ts new file mode 100644 index 0000000..85f05d6 --- /dev/null +++ b/src/mgr/abortable-child.ts @@ -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 { + const killGraceMs = input.killGraceMs ?? childProcessKillGraceMs; + return new Promise((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 } }); +} diff --git a/src/mgr/kafka-outbox-relay.ts b/src/mgr/kafka-outbox-relay.ts index 9557ed6..875d02a 100644 --- a/src/mgr/kafka-outbox-relay.ts +++ b/src/mgr/kafka-outbox-relay.ts @@ -14,17 +14,18 @@ export interface KafkaOutboxRelayOptions { } export async function relayKafkaEventOutboxOnce(input: { store: AgentRunStore; options: KafkaOutboxRelayOptions; publish?: (config: AgentRunKafkaConfig, message: AgentRunKafkaMessage) => Promise }): Promise { - 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(); 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 { @@ -81,8 +83,14 @@ export function startKafkaEventOutboxRelay(input: { store: AgentRunStore; option }; } -async function retry(store: AgentRunStore, options: KafkaOutboxRelayOptions, item: KafkaEventOutboxRecord, error: JsonRecord): Promise { - 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 { + 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 { diff --git a/src/mgr/kubernetes-runner-job.ts b/src/mgr/kubernetes-runner-job.ts index d2fb8f3..88bd05e 100644 --- a/src/mgr/kubernetes-runner-job.ts +++ b/src/mgr/kubernetes-runner-job.ts @@ -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 { +export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise { + 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 { - await kubectlRun(kubectlCommand, ["delete", "secret", name, "-n", namespace, "--ignore-not-found=true"]); +async function kubectlDeleteSecret(name: string, namespace: string, kubectlCommand: string, abortSignal?: AbortSignal): Promise { + 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 }; } diff --git a/src/mgr/postgres-durable-queues.ts b/src/mgr/postgres-durable-queues.ts index 2c118da..0ba8d30 100644 --- a/src/mgr/postgres-durable-queues.ts +++ b/src/mgr/postgres-durable-queues.ts @@ -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), diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index ceed960..514e307 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -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 { - 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 { - 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 { + async completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): Promise { 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 { + 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 { + 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 { + async completeKafkaEventOutbox(claim: DurableQueueClaim): Promise { 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 { + async retryKafkaEventOutbox(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string): Promise { 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), diff --git a/src/mgr/runner-dispatch-intent.ts b/src/mgr/runner-dispatch-intent.ts index bddf618..b1c0e18 100644 --- a/src/mgr/runner-dispatch-intent.ts +++ b/src/mgr/runner-dispatch-intent.ts @@ -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, diff --git a/src/mgr/runner-dispatcher.ts b/src/mgr/runner-dispatcher.ts index b2d3a64..47e4fac 100644 --- a/src/mgr/runner-dispatcher.ts +++ b/src/mgr/runner-dispatcher.ts @@ -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 { - 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 => { 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 { + 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): 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(intent: RunnerDispatchIntentRecord, timeoutMs: number, parentSignal: AbortSignal | undefined, callback: (signal: AbortSignal) => Promise): Promise { + 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); + } +} diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index e983918..9aa280d 100644 --- a/src/mgr/runner-retention.ts +++ b/src/mgr/runner-retention.ts @@ -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 { - 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).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 }; } diff --git a/src/mgr/server.ts b/src/mgr/server.ts index a21a703..c6a0b6e 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -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); diff --git a/src/mgr/session-pvc.ts b/src/mgr/session-pvc.ts index 8433231..b224a2a 100644 --- a/src/mgr/session-pvc.ts +++ b/src/mgr/session-pvc.ts @@ -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 { @@ -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); diff --git a/src/mgr/store.ts b/src/mgr/store.ts index c5faaf0..e5311e7 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -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; createRun(input: CreateRunInput): MaybePromise; @@ -37,13 +43,13 @@ export interface AgentRunStore { listCommands(runId: string, afterSeq: number, limit: number): MaybePromise; getRunnerDispatchIntent(commandId: string): MaybePromise; claimRunnerDispatchIntents(input: { owner: string; leaseMs: number; limit: number }): MaybePromise; - completeRunnerDispatchIntent(intentId: string, runnerJobId: string): MaybePromise; - retryRunnerDispatchIntent(intentId: string, error: JsonRecord, nextAttemptAt: string): MaybePromise; - terminalizeRunnerDispatchIntent(intentId: string, error: JsonRecord): MaybePromise; + completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): MaybePromise; + retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): MaybePromise; + terminalizeRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord): MaybePromise; runnerDispatchStatus(): MaybePromise; claimKafkaEventOutbox(input: { owner: string; leaseMs: number; limit: number }): MaybePromise; - completeKafkaEventOutbox(outboxId: string): MaybePromise; - retryKafkaEventOutbox(outboxId: string, error: JsonRecord, nextAttemptAt: string): MaybePromise; + completeKafkaEventOutbox(claim: DurableQueueClaim): MaybePromise; + retryKafkaEventOutbox(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string): MaybePromise; kafkaEventOutboxStatus(): MaybePromise; registerRunner(input: Partial): MaybePromise; listRunnerJobs(runId: string, commandId?: string): MaybePromise; @@ -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): RunRecord { @@ -1154,6 +1186,13 @@ function commandForEvent(commands: Map, 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(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); diff --git a/src/selftest/cases/00-redaction-postgres.ts b/src/selftest/cases/00-redaction-postgres.ts index d538282..15f4fc6 100644 --- a/src/selftest/cases/00-redaction-postgres.ts +++ b/src/selftest/cases/00-redaction-postgres.ts @@ -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)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record)["008_v01_dsflash_go_backend_profile"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record)["009_v01_dsflash_go_model_catalog"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record)["010_v01_queue_session_ref"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["011_v01_cancel_lifecycle"] === "string" && (postgresContract.checksums as Record)["011_v01_cancel_lifecycle"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"] === "string" && (postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"].length > 0); + assert.ok(typeof (postgresContract.checksums as Record)["013_v02_runner_dispatch_outcome"] === "string" && (postgresContract.checksums as Record)["013_v02_runner_dispatch_outcome"].length > 0); assert.equal((postgresContract.checksums as Record)["002_v01_backend_profiles"], "928b5c490cc4539cb64ecef34784557601b2724fa2870570f16a53576804e49c"); assert.ok(Array.isArray(postgresContract.requiredTables)); assert.ok(postgresContract.requiredTables.includes("agentrun_schema_migrations")); diff --git a/src/selftest/cases/21-runner-durable-dispatch.ts b/src/selftest/cases/21-runner-durable-dispatch.ts index 5d34c7f..35c6bf3 100644 --- a/src/selftest/cases/21-runner-durable-dispatch.ts +++ b/src/selftest/cases/21-runner-durable-dispatch.ts @@ -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 { @@ -81,7 +86,7 @@ async function assertHttpAdmissionContract(kubectlCommand: string): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { + 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; diff --git a/src/selftest/cases/35-kafka-durable-outbox.ts b/src/selftest/cases/35-kafka-durable-outbox.ts index 8f4760f..813aaa1 100644 --- a/src/selftest/cases/35-kafka-durable-outbox.ts +++ b/src/selftest/cases/35-kafka-durable-outbox.ts @@ -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 => { 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 { + 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 { 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 { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + export default selfTest; diff --git a/src/selftest/fixtures/fake-kubectl-adopt.mjs b/src/selftest/fixtures/fake-kubectl-adopt.mjs index 519cb79..217bfd6 100755 --- a/src/selftest/fixtures/fake-kubectl-adopt.mjs +++ b/src/selftest/fixtures/fake-kubectl-adopt.mjs @@ -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);