From 52ff8a54548455748474e595045a6728bacd98bf Mon Sep 17 00:00:00 2001 From: root Date: Fri, 10 Jul 2026 03:50:42 +0200 Subject: [PATCH] fix: serialize Kafka outbox and own runner runtime config --- src/common/validation.ts | 79 ++++--------------- src/mgr/event-outbox.ts | 6 +- src/mgr/kubernetes-runner-job.ts | 3 +- src/mgr/postgres-store.ts | 17 ++-- .../cases/21-runner-durable-dispatch.ts | 16 ++-- src/selftest/cases/35-kafka-durable-outbox.ts | 26 +++--- 6 files changed, 56 insertions(+), 91 deletions(-) diff --git a/src/common/validation.ts b/src/common/validation.ts index e411a6c..6304a52 100644 --- a/src/common/validation.ts +++ b/src/common/validation.ts @@ -2,7 +2,6 @@ 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); @@ -333,42 +332,11 @@ export function validateCreateCommand(input: unknown): CreateCommandInput { 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 allowedKeys = new Set(["idempotencyKey", "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 } }); + if (rejectedKeys.length > 0) throw new AgentRunError("tenant-policy-denied", "durable dispatch runtime settings are controlled by AgentRun configuration", { httpStatus: 403, 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.idempotencyKey !== undefined) normalized.idempotencyKey = boundedDispatchString(input.idempotencyKey, "command.dispatch.input.idempotencyKey", 256); if (input.transientEnv !== undefined) normalized.transientEnv = validateDurableTransientEnv(input.transientEnv); return normalized; } @@ -383,6 +351,7 @@ function validateDurableTransientEnv(value: unknown): JsonRecord[] { 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`); + assertRunnerTransientEnvNameAllowed(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); @@ -392,6 +361,18 @@ function validateDurableTransientEnv(value: unknown): JsonRecord[] { }); } +export function assertRunnerTransientEnvNameAllowed(name: string): void { + const normalized = name.toUpperCase(); + const reservedPrefixes = ["AGENTRUN_", "CODEX_", "GIT_", "KUBERNETES_", "NPM_CONFIG_", "OTEL_"]; + const reservedNames = new Set([ + "ALL_PROXY", "BASH_ENV", "BUN_OPTIONS", "CODEX_HOME", "ENV", "HOME", "HTTP_PROXY", "HTTPS_PROXY", + "KUBECONFIG", "LD_LIBRARY_PATH", "LD_PRELOAD", "NODE_EXTRA_CA_CERTS", "NODE_OPTIONS", "NO_PROXY", "PATH", "PROMPT_COMMAND", + ]); + if (reservedNames.has(normalized) || reservedPrefixes.some((prefix) => normalized.startsWith(prefix))) { + throw new AgentRunError("tenant-policy-denied", `transientEnv ${name} is controlled by AgentRun runtime configuration`, { httpStatus: 403, details: { name, valuesPrinted: false } }); + } +} + function assertNoSecretNamedFields(value: unknown, path: string): void { if (Array.isArray(value)) { value.forEach((entry, index) => assertNoSecretNamedFields(entry, `${path}[${index}]`)); @@ -428,34 +409,6 @@ function boundedDispatchString(value: unknown, fieldName: string, maxBytes: numb 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; diff --git a/src/mgr/event-outbox.ts b/src/mgr/event-outbox.ts index ad25024..f84fdd8 100644 --- a/src/mgr/event-outbox.ts +++ b/src/mgr/event-outbox.ts @@ -8,6 +8,10 @@ export interface CanonicalAgentRunEventMessage { headers: JsonRecord; } +export function canonicalAgentRunEventPartitionKey(run: RunRecord): string { + return text(run.sessionRef?.sessionId) ?? run.id; +} + export function buildCanonicalAgentRunEventMessage(input: { source: string; topic: string; @@ -75,7 +79,7 @@ export function buildCanonicalAgentRunEventMessage(input: { }; return { topic: input.topic, - key: sessionId ?? run.id, + key: canonicalAgentRunEventPartitionKey(run), value, headers: { "x-event-schema": "agentrun.event.v1", diff --git a/src/mgr/kubernetes-runner-job.ts b/src/mgr/kubernetes-runner-job.ts index 88bd05e..44ed418 100644 --- a/src/mgr/kubernetes-runner-job.ts +++ b/src/mgr/kubernetes-runner-job.ts @@ -4,7 +4,7 @@ import { redactJson, redactText } from "../common/redaction.js"; import { isLeaseExpired, isTerminalCommandState, isTerminalRunStatus, summarizeResourceBundleRef, summarizeSessionRef } from "./store.js"; import type { AgentRunStore } from "./store.js"; import type { ExecutionPolicy, JsonRecord } from "../common/types.js"; -import { stableHash, validateEnvName } from "../common/validation.js"; +import { assertRunnerTransientEnvNameAllowed, stableHash, validateEnvName } from "../common/validation.js"; import { renderRunnerJobManifest } from "../runner/k8s-job.js"; import type { RunnerSessionPvcOptions, RunnerTransientEnv } from "../runner/k8s-job.js"; import { staticWorkReadyCapabilitySummary } from "../common/work-ready.js"; @@ -364,6 +364,7 @@ function transientEnvField(value: unknown): RunnerTransientEnv[] { const record = entry as JsonRecord; const name = stringField(record, "name"); validateEnvName(name, `transientEnv[${index}].name`); + assertRunnerTransientEnvNameAllowed(name); if (reusableCredentialEnvNames.has(name)) throw new AgentRunError("tenant-policy-denied", `transientEnv ${name} must use tool/provider credential assembly instead`, { httpStatus: 403 }); if (seen.has(name)) throw new AgentRunError("schema-invalid", `transientEnv name ${name} is duplicated`, { httpStatus: 400 }); seen.add(name); diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 514e307..ae82743 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -9,7 +9,7 @@ import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTask 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 { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js"; import { assertRunnerDispatchReplay, attachRunnerDispatchIntent, newRunnerDispatchIntent } from "./runner-dispatch-intent.js"; import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, runnerDispatchIntentFromRow, runnerDispatchOutcomeMigrationSql } from "./postgres-durable-queues.js"; @@ -1414,6 +1414,12 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( const seq = await this.nextSeq(client, "agentrun_events", runId); const event: RunEvent = { id: newId("evt"), runId, seq, type: fenced.type, payload: redactJson(eventPayload), createdAt: nowIso() }; await client.query("INSERT INTO agentrun_events (id, run_id, seq, type, payload, created_at) VALUES ($1, $2, $3, $4, $5::jsonb, $6)", [event.id, event.runId, event.seq, event.type, JSON.stringify(event.payload), event.createdAt]); + await client.query( + `UPDATE agentrun_sessions + SET last_event_seq = $2, last_activity_at = $3, updated_at = $3 + WHERE session_id = (SELECT session_ref->>'sessionId' FROM agentrun_runs WHERE id = $1)`, + [runId, event.seq, event.createdAt], + ); if (this.eventOutboxConfig.enabled) { const commandId = firstString(event.payload.commandId, event.payload.targetCommandId); let command: CommandRecord | null = null; @@ -1421,6 +1427,9 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 AND run_id = $2", [commandId, runId]); if (commandResult.rows[0]) command = commandFromRow(commandResult.rows[0]); } + const outboxTopic = this.eventOutboxConfig.topic as string; + const partitionKey = canonicalAgentRunEventPartitionKey(run); + await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", [outboxTopic, partitionKey]); const outboxSeqResult = await client.query<{ value: string | number }>("SELECT nextval('agentrun_kafka_event_outbox_seq') AS value"); const outbox = buildKafkaEventOutboxRecord({ source: this.eventOutboxConfig.source as string, topic: this.eventOutboxConfig.topic as string, outboxSeq: Number(outboxSeqResult.rows[0]?.value), run, command, event }); await client.query( @@ -1429,12 +1438,6 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( [outbox.id, outbox.outboxSeq, outbox.eventId, outbox.runId, outbox.sourceSeq, outbox.topic, outbox.partitionKey, JSON.stringify(outbox.value), JSON.stringify(outbox.headers), outbox.attemptCount, outbox.leaseOwner, outbox.leaseExpiresAt, outbox.nextAttemptAt, JSON.stringify(outbox.lastError), outbox.deliveredAt, outbox.createdAt, outbox.updatedAt], ); } - await client.query( - `UPDATE agentrun_sessions - SET last_event_seq = $2, last_activity_at = $3, updated_at = $3 - WHERE session_id = (SELECT session_ref->>'sessionId' FROM agentrun_runs WHERE id = $1)`, - [runId, event.seq, event.createdAt], - ); return event; } diff --git a/src/selftest/cases/21-runner-durable-dispatch.ts b/src/selftest/cases/21-runner-durable-dispatch.ts index 35c6bf3..ec9a8e3 100644 --- a/src/selftest/cases/21-runner-durable-dispatch.ts +++ b/src/selftest/cases/21-runner-durable-dispatch.ts @@ -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, namespace: "agentrun-v02", serviceAccountName: "agentrun-v02-runner", transientEnv: hwlabRuntimeEnv() } }, + dispatch: { kind: "kubernetes-job", input: { 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) })); @@ -54,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, namespace: "agentrun-v02", serviceAccountName: "agentrun-v02-runner", transientEnv: hwlabRuntimeEnv() } }, + dispatch: { kind: "kubernetes-job", input: { transientEnv: hwlabRuntimeEnv() } }, })); const failed = await dispatchRunnerIntentsOnce({ store, defaults: defaults(fakeKubectl), options: dispatcherOptions(1) }); assert.equal(failed.failedCount, 1); @@ -86,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 } } })); + const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); store.claimRun(run.id, "runner_existing", 60_000); const result = await dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options: dispatcherOptions(3) }); assert.equal(result.dispatchedCount, 1); @@ -134,7 +134,7 @@ async function assertActiveRunnerReuseHasNoDanglingJob(kubectlCommand: string): 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 } } })); + const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "hang"; const marker = `${process.env.AGENTRUN_SELFTEST_KUBECTL_STATE}.hang-started`; await rm(marker, { force: true }); @@ -154,9 +154,9 @@ async function assertHungKubectlTimesOutAndRecovers(kubectlCommand: string): Pro 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 first = store.createCommand(firstRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); const secondRun = store.createRun(runInput()); - const second = store.createCommand(secondRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } })); + const second = store.createCommand(secondRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "hang"; const marker = `${process.env.AGENTRUN_SELFTEST_KUBECTL_STATE}.hang-started`; await rm(marker, { force: true }); @@ -175,7 +175,7 @@ async function assertClaimsAreLeasedJustInTime(kubectlCommand: string): Promise< 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 } } })); + const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } })); 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"); diff --git a/src/selftest/cases/35-kafka-durable-outbox.ts b/src/selftest/cases/35-kafka-durable-outbox.ts index 813aaa1..5809fd7 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: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", dockerfilePath: "Containerfile" } }, + input: {}, }, }); const command = store.createCommand(run.id, commandInput); @@ -67,23 +67,27 @@ const selfTest: SelfTestCase = async () => { (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", + () => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { managerUrl: "https://manager.example.test" } } }), + (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403, ); 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", + () => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { namespace: "attacker-runtime" } } }), + (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403, ); 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", + () => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image: "attacker.example.test/runner:latest" } } }), + (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403, ); 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", + () => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { serviceAccountName: "cluster-admin" } } }), + (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403, ); - 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 }] }); + assert.throws( + () => validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { transientEnv: [{ name: "AGENTRUN_MGR_URL", value: "https://attacker.example.test", sensitive: false }] } } }), + (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403, + ); + const normalizedDispatch = validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { transientEnv: [{ name: "HWLAB_RUNTIME_LANE", value: "v02", sensitive: false }] } } }).dispatch?.input; + assert.deepEqual(normalizedDispatch, { 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);