fix: serialize Kafka outbox and own runner runtime config
This commit is contained in:
root
+16
-63
@@ -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;
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void
|
||||
type: "turn",
|
||||
idempotencyKey: "http-durable-admission",
|
||||
payload: { prompt: "durable HTTP admission" },
|
||||
dispatch: { kind: "kubernetes-job", input: { image, namespace: "agentrun-v02", serviceAccountName: "agentrun-v02-runner", transientEnv: hwlabRuntimeEnv() } },
|
||||
dispatch: { kind: "kubernetes-job", input: { transientEnv: hwlabRuntimeEnv() } },
|
||||
}) as { dispatchIntent?: { id?: string; state?: string; runnerJobId?: string; attemptCount?: number; durable?: boolean }; dispatch?: unknown };
|
||||
assert.match(command.dispatchIntent?.id ?? "", /^dispatch_[a-f0-9]{24}$/u);
|
||||
assert.equal(command.dispatchIntent?.state, "pending");
|
||||
@@ -119,7 +119,7 @@ function dispatcherOptions(maxAttempts: number): RunnerDispatcherOptions {
|
||||
async function assertActiveRunnerReuseHasNoDanglingJob(kubectlCommand: string): Promise<void> {
|
||||
const store = new MemoryAgentRunStore();
|
||||
const run = store.createRun(runInput());
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
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<void> {
|
||||
const store = new MemoryAgentRunStore();
|
||||
const run = store.createRun(runInput());
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
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<void> {
|
||||
const store = new MemoryAgentRunStore();
|
||||
const firstRun = store.createRun(runInput());
|
||||
const first = store.createCommand(firstRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
const 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<void> {
|
||||
const store = new FailingDispatchEventStore({ eventOutbox: { enabled: true, topic: "agentrun.event.v1", source: "selftest" } });
|
||||
const run = store.createRun(runInput());
|
||||
const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } }));
|
||||
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");
|
||||
|
||||
@@ -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);
|
||||
|
||||
Reference in New Issue
Block a user