From 51aaafd21d510f483eacc23f6cca3e050d2db386 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 10 Jul 2026 03:47:14 +0200 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20durable=20dispatch=20?= =?UTF-8?q?=E7=BB=88=E6=80=81=E9=9A=94=E7=A6=BB=E4=B8=8E=20runner=20?= =?UTF-8?q?=E5=87=AD=E6=8D=AE=E5=BC=95=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/kubernetes-runner-job.ts | 6 +- src/mgr/postgres-runner-dispatch.ts | 65 ++++++++++ src/mgr/postgres-store.ts | 118 ++++++++++-------- src/mgr/provider-profiles.ts | 4 + src/mgr/server.ts | 8 ++ src/mgr/store.ts | 62 ++++++++- src/runner/k8s-job.ts | 21 +++- src/selftest/cases/20-runner-k8s-job.ts | 15 +++ .../cases/21-runner-durable-dispatch.ts | 63 +++++++++- .../cases/45-provider-profile-management.ts | 1 + .../cases/50-hwlab-manual-dispatch.ts | 1 + .../cases/60-hwlab-baseline-contract.ts | 1 + src/selftest/cases/75-queue-q2-dispatch.ts | 1 + 13 files changed, 304 insertions(+), 62 deletions(-) create mode 100644 src/mgr/postgres-runner-dispatch.ts diff --git a/src/mgr/kubernetes-runner-job.ts b/src/mgr/kubernetes-runner-job.ts index 44ed418..facb6ea 100644 --- a/src/mgr/kubernetes-runner-job.ts +++ b/src/mgr/kubernetes-runner-job.ts @@ -6,7 +6,7 @@ import type { AgentRunStore } from "./store.js"; import type { ExecutionPolicy, JsonRecord } from "../common/types.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 type { RunnerApiKeySecretRef, RunnerSessionPvcOptions, RunnerTransientEnv } from "../runner/k8s-job.js"; import { staticWorkReadyCapabilitySummary } from "../common/work-ready.js"; import { resolveRunnerEnvImage } from "../common/env-image-ref.js"; import { ensureSessionPvc } from "./session-pvc.js"; @@ -41,6 +41,7 @@ const unideskSshEndpointConfigEnvNames: Array<{ configName: string; targetName: export interface RunnerJobDefaults { namespace: string; managerUrl: string; + runnerApiKeySecretRef: RunnerApiKeySecretRef; image: string; bootRepoUrl?: string; sourceCommit: string; @@ -117,6 +118,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore; image, namespace, managerUrl, + runnerApiKeySecretRef: options.defaults.runnerApiKeySecretRef, sourceCommit, envImage, serviceAccountName: serviceAccountName ?? null, @@ -209,6 +211,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore; run, commandId, managerUrl, + runnerApiKeySecretRef: options.defaults.runnerApiKeySecretRef, image, ...(options.defaults.bootRepoUrl ? { bootRepoUrl: options.defaults.bootRepoUrl } : {}), namespace, @@ -283,6 +286,7 @@ export async function createKubernetesRunnerJob(options: { store: AgentRunStore; logPath: `kubectl -n ${render.namespace} logs job/${render.jobName}`, }, envImage, + runnerApiKeySecretRef: { ...options.defaults.runnerApiKeySecretRef, valuesPrinted: false }, secretRefs: render.secretRefs.map((item) => ({ profile: item.profile, name: item.secretRef.name, namespace: item.secretRef.namespace ?? render.namespace, keys: item.secretRef.keys ?? [], mountPath: item.runtimeMountPath, projectionPath: item.projectionMountPath, writableCopy: true, valuesPrinted: false })), toolCredentials: summarizeToolCredentials(render.toolCredentials, render.namespace), gitTransport: gitTransportSummary(), diff --git a/src/mgr/postgres-runner-dispatch.ts b/src/mgr/postgres-runner-dispatch.ts new file mode 100644 index 0000000..c0da1dc --- /dev/null +++ b/src/mgr/postgres-runner-dispatch.ts @@ -0,0 +1,65 @@ +import type { PoolClient, QueryResultRow } from "pg"; +import { AgentRunError } from "../common/errors.js"; +import { nowIso } from "../common/validation.js"; +import type { DurableQueueClaim } from "./store.js"; + +export interface LockedRunnerDispatchClaim { + intentRow: QueryResultRow; + commandRow: QueryResultRow; + runRow: QueryResultRow; + fenced: boolean; +} + +export async function lockRunnerDispatchClaim(client: PoolClient, claim: DurableQueueClaim): Promise { + const lookup = await client.query("SELECT run_id, command_id FROM agentrun_runner_dispatch_intents WHERE id = $1", [claim.id]); + const identity = lookup.rows[0]; + if (!identity) throw staleClaimError("runner dispatch", claim); + const runResult = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [identity.run_id]); + if (!runResult.rows[0]) throw new AgentRunError("schema-invalid", `run ${String(identity.run_id)} was not found`, { httpStatus: 404 }); + const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [identity.run_id]); + const commandRow = commandResult.rows.find((row) => String(row.id) === String(identity.command_id)); + if (!commandRow) throw new AgentRunError("schema-invalid", `command ${String(identity.command_id)} was not found`, { httpStatus: 404 }); + 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 runRow = runResult.rows[0]; + if (!terminalRunStatuses.has(String(runRow.status)) && !terminalCommandStates.has(String(commandRow.state))) return { intentRow: intentResult.rows[0], commandRow, runRow, fenced: false }; + const reason = terminalRunStatuses.has(String(runRow.status)) ? `run is terminal: ${String(runRow.status)}` : `command is terminal: ${String(commandRow.state)}`; + const fenced = await client.query( + "UPDATE agentrun_runner_dispatch_intents SET state = 'cancelled', lease_owner = NULL, lease_expires_at = NULL, last_error = $2::jsonb, updated_at = $3 WHERE id = $1 RETURNING *", + [claim.id, JSON.stringify({ code: "cancelled", message: reason, valuesPrinted: false }), nowIso()], + ); + return { intentRow: fenced.rows[0], commandRow, runRow, fenced: true }; +} + +export async function fenceActiveRunnerDispatchIntents(client: PoolClient, selector: { runId: string } | { commandId: string }, reason: string, at = nowIso()): Promise { + const byRun = "runId" in selector; + const result = await client.query( + `UPDATE agentrun_runner_dispatch_intents + SET state = 'cancelled', lease_owner = NULL, lease_expires_at = NULL, last_error = $2::jsonb, updated_at = $3 + WHERE ${byRun ? "run_id" : "command_id"} = $1 AND state IN ('pending', 'dispatching', 'retry')`, + [byRun ? selector.runId : selector.commandId, JSON.stringify({ code: "cancelled", message: reason, valuesPrinted: false }), at], + ); + return result.rowCount ?? 0; +} + +export async function fenceTerminalRunnerDispatchIntents(client: PoolClient): Promise { + const result = await client.query( + `UPDATE agentrun_runner_dispatch_intents i + SET state = 'cancelled', lease_owner = NULL, lease_expires_at = NULL, + last_error = '{"code":"cancelled","message":"run or command is terminal","valuesPrinted":false}'::jsonb, updated_at = now() + FROM agentrun_runs r, agentrun_commands c + WHERE i.run_id = r.id AND i.command_id = c.id AND i.state IN ('pending', 'dispatching', 'retry') + AND (r.status IN ('completed', 'failed', 'blocked', 'cancelled') OR c.state IN ('completed', 'failed', 'cancelled'))`, + ); + return result.rowCount ?? 0; +} + +export 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 } }); +} + +const terminalRunStatuses = new Set(["completed", "failed", "blocked", "cancelled"]); +const terminalCommandStates = new Set(["completed", "failed", "cancelled"]); diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index ae82743..68995ff 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -12,6 +12,7 @@ import { normalizeRunEventPayload, requireEventType } from "../common/events.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"; +import { fenceActiveRunnerDispatchIntents, fenceTerminalRunnerDispatchIntents, lockRunnerDispatchClaim, staleClaimError } from "./postgres-runner-dispatch.js"; interface PostgresStoreOptions { connectionString: string; @@ -608,69 +609,70 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( async claimRunnerDispatchIntents(input: { owner: string; leaseMs: number; limit: number }): Promise { const leaseExpiresAt = new Date(Date.now() + input.leaseMs).toISOString(); - const result = await this.pool.query( - `WITH due AS ( - SELECT id FROM agentrun_runner_dispatch_intents - WHERE next_attempt_at <= now() - AND (state IN ('pending', 'retry') OR (state = 'dispatching' AND (lease_expires_at IS NULL OR lease_expires_at <= now()))) - ORDER BY next_attempt_at ASC, created_at ASC - FOR UPDATE SKIP LOCKED - LIMIT $1 - ) - UPDATE agentrun_runner_dispatch_intents i - SET state = 'dispatching', attempt_count = i.attempt_count + 1, lease_owner = $2, lease_expires_at = $3, updated_at = now() - FROM due WHERE i.id = due.id - RETURNING i.*`, - [clamp(input.limit, 1, 500), input.owner, leaseExpiresAt], - ); - return result.rows.map(runnerDispatchIntentFromRow); + return this.withTransaction(async (client) => { + await fenceTerminalRunnerDispatchIntents(client); + const result = await client.query( + `WITH due AS ( + SELECT i.id FROM agentrun_runner_dispatch_intents i + JOIN agentrun_runs r ON r.id = i.run_id JOIN agentrun_commands c ON c.id = i.command_id + WHERE i.next_attempt_at <= now() + AND r.status NOT IN ('completed', 'failed', 'blocked', 'cancelled') AND c.state NOT IN ('completed', 'failed', 'cancelled') + AND (i.state IN ('pending', 'retry') OR (i.state = 'dispatching' AND (i.lease_expires_at IS NULL OR i.lease_expires_at <= now()))) + ORDER BY i.next_attempt_at ASC, i.created_at ASC FOR UPDATE OF i SKIP LOCKED LIMIT $1 + ) + UPDATE agentrun_runner_dispatch_intents i + SET state = 'dispatching', attempt_count = i.attempt_count + 1, lease_owner = $2, lease_expires_at = $3, updated_at = now() + FROM due WHERE i.id = due.id RETURNING i.*`, + [clamp(input.limit, 1, 500), input.owner, leaseExpiresAt], + ); + return result.rows.map(runnerDispatchIntentFromRow); + }); } async completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): Promise { - return this.withTransaction(async (client) => { + const settled = await this.withTransaction(async (client) => { + const locked = await lockRunnerDispatchClaim(client, claim); + if (locked.fenced) return runnerDispatchIntentFromRow(locked.intentRow); 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], + WHERE id = $1 RETURNING *`, + [claim.id, completion.dispatchOutcome, completion.actualRunnerJobId, completion.activeRunnerId], ); - 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; }); + if (settled.state === "cancelled") throw staleClaimError("runner dispatch", claim); + return settled; } async retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): Promise { - return this.withTransaction(async (client) => { + const settled = await this.withTransaction(async (client) => { + const locked = await lockRunnerDispatchClaim(client, claim); + if (locked.fenced) return runnerDispatchIntentFromRow(locked.intentRow); 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], + WHERE id = $1 RETURNING *`, + [claim.id, nextAttemptAt, JSON.stringify(redactJson(error))], ); - 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; }); + if (settled.state === "cancelled") throw staleClaimError("runner dispatch", claim); + return settled; } 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 }); - const command = commandFromRow(commandResult.rows[0]); - const run = await this.requireRunForUpdate(client, intent.runId); + const settled = await this.withTransaction(async (client) => { + const locked = await lockRunnerDispatchClaim(client, claim); + const intent = runnerDispatchIntentFromRow(locked.intentRow); + const command = commandFromRow(locked.commandRow); + const run = runFromRow(locked.runRow); + if (locked.fenced) return { intent, command, run, dispatchFenced: true, valuesPrinted: false }; const failure = redactJson(error); const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed"; const at = nowIso(); @@ -694,6 +696,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( [run.id, message, at], ); updatedRun = runFromRow(updated.rows[0]); + await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, "run terminalized by runner dispatch failure", at); await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "runner-dispatch-failed", commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message, threadId: null, turnId: null }); await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message }); @@ -701,6 +704,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( } return { intent: runnerDispatchIntentFromRow(updatedIntentResult.rows[0]), command: updatedCommand, run: updatedRun, valuesPrinted: false }; }); + if (settled.dispatchFenced === true) throw staleClaimError("runner dispatch", claim); + return settled; } async runnerDispatchStatus(): Promise { @@ -924,20 +929,24 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( async finishCommand(commandId: string, result: Pick): Promise { return this.withTransaction(async (client) => { + const identity = await client.query("SELECT run_id FROM agentrun_commands WHERE id = $1", [commandId]); + if (!identity.rows[0]) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 }); + const run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id)); const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]); const row = existing.rows[0]; if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 }); const command = commandFromRow(row); if (isTerminalCommandState(command.state)) { + await fenceActiveRunnerDispatchIntents(client, { commandId }, "command is terminal"); if (command.state === "cancelled" && result.terminalStatus !== "cancelled") { - const run = await this.requireRunForUpdate(client, command.runId); await this.appendEventWithLockedRun(client, command.runId, "backend_status", lateWriteRejectedPayload(run, command, { source: "command-status", terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null })); } return command; } const state = commandStateFromTerminal(result.terminalStatus); - const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, nowIso()]); - const run = await this.requireRunForUpdate(client, command.runId); + const at = nowIso(); + const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, at]); + await fenceActiveRunnerDispatchIntents(client, { commandId }, `command terminalized as ${state}`, at); if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null); if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: nowIso() }, { bumpVersion: true }); await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state, terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null }); @@ -955,16 +964,20 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( async finishRun(runId: string, result: Pick): Promise { return this.withTransaction(async (client) => { const existing = await this.requireRunForUpdate(client, runId); + await client.query("SELECT id FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [runId]); if (isTerminalRunStatus(existing.status)) { + await fenceActiveRunnerDispatchIntents(client, { runId }, "run is terminal"); if (existing.status === "cancelled" && result.terminalStatus !== "cancelled") await this.appendEventWithLockedRun(client, runId, "backend_status", lateWriteRejectedPayload(existing, null, { source: "run-status", terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null })); return existing; } const status = statusFromTerminal(result.terminalStatus); + const at = nowIso(); const updated = await client.query( `UPDATE agentrun_runs SET status = $2, terminal_status = $3, failure_kind = $4, failure_message = $5, updated_at = $6 WHERE id = $1 RETURNING *`, - [runId, status, result.terminalStatus, result.failureKind, result.failureMessage, nowIso()], + [runId, status, result.terminalStatus, result.failureKind, result.failureMessage, at], ); const run = runFromRow(updated.rows[0]); + await fenceActiveRunnerDispatchIntents(client, { runId }, `run terminalized as ${status}`, at); if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null); await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage }); await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: run.updatedAt }, { bumpVersion: true, at: run.updatedAt }); @@ -975,7 +988,11 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( async cancelRun(runId: string, reason = "cancel requested"): Promise { return this.withTransaction(async (client) => { const run = await this.requireRunForUpdate(client, runId); - if (isTerminalRunStatus(run.status)) return run; + const commands = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [runId]); + if (isTerminalRunStatus(run.status)) { + await fenceActiveRunnerDispatchIntents(client, { runId }, reason); + return run; + } const at = nowIso(); const cancel = await this.createCancelRequest(client, { targetKind: "run", targetId: runId, run, command: null, reason, at, stage: "accepted" }); await this.appendCancelStage(client, runId, "accepted", cancel); @@ -991,12 +1008,13 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( await this.appendCancelStage(client, runId, "delivered", cancel, { claimedBy: persistedRun.claimedBy }); await this.appendCancelStage(client, runId, "aborting", cancel, { claimedBy: persistedRun.claimedBy }); } - const commands = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND state NOT IN ('completed', 'failed', 'cancelled') FOR UPDATE", [runId]); for (const row of commands.rows) { const command = commandFromRow(row); + if (isTerminalCommandState(command.state)) continue; await client.query("UPDATE agentrun_commands SET state = 'cancelled', cancel_epoch = $2, cancel_request_id = $3, cancel_requested_at = $4, cancel_reason = $5, updated_at = $4 WHERE id = $1", [command.id, cancel.epoch, cancel.id, at, reason]); await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-terminal", commandId: command.id, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", cancelRequestId: cancel.id, cancelEpoch: cancel.epoch }); } + await fenceActiveRunnerDispatchIntents(client, { runId }, reason, at); const updated = await client.query( `UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, updated_at = $3 WHERE id = $1 RETURNING *`, [runId, reason, at], @@ -1011,16 +1029,22 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( async cancelCommand(commandId: string, reason = "cancel requested"): Promise { return this.withTransaction(async (client) => { + const identity = await client.query("SELECT run_id FROM agentrun_commands WHERE id = $1", [commandId]); + if (!identity.rows[0]) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 }); + const run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id)); const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]); const row = existing.rows[0]; if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 }); const command = commandFromRow(row); - if (isTerminalCommandState(command.state)) return command; - const run = await this.requireRunForUpdate(client, command.runId); + if (isTerminalCommandState(command.state)) { + await fenceActiveRunnerDispatchIntents(client, { commandId }, reason); + return command; + } const at = nowIso(); const cancel = await this.createCancelRequest(client, { targetKind: "command", targetId: commandId, run, command, reason, at, stage: "accepted" }); await this.appendCancelStage(client, command.runId, "accepted", cancel); const updated = await client.query("UPDATE agentrun_commands SET state = 'cancelled', cancel_epoch = $2, cancel_request_id = $3, cancel_requested_at = $4, cancel_reason = $5, updated_at = $4 WHERE id = $1 RETURNING *", [commandId, cancel.epoch, cancel.id, at, reason]); + await fenceActiveRunnerDispatchIntents(client, { commandId }, reason, at); await this.appendCancelStage(client, command.runId, "persisted", cancel); const leaseExpired = Boolean(run.claimedBy && isLeaseExpired(run.leaseExpiresAt)); if (leaseExpired) await this.appendCancelStage(client, command.runId, "fenced", cancel, { claimedBy: run.claimedBy, leaseExpiresAt: run.leaseExpiresAt }); @@ -1678,10 +1702,6 @@ 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/provider-profiles.ts b/src/mgr/provider-profiles.ts index 94cf23f..dd9a254 100644 --- a/src/mgr/provider-profiles.ts +++ b/src/mgr/provider-profiles.ts @@ -682,6 +682,10 @@ function runnerDefaultsForValidation(options: ProviderProfileValidationOptions, return { namespace, managerUrl: options.runnerJobDefaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL ?? `http://agentrun-mgr.${namespace}.svc.cluster.local:8080`, + runnerApiKeySecretRef: options.runnerJobDefaults?.runnerApiKeySecretRef ?? { + name: process.env.AGENTRUN_RUNNER_API_KEY_SECRET_NAME ?? "", + key: process.env.AGENTRUN_RUNNER_API_KEY_SECRET_KEY ?? "", + }, image: options.runnerJobDefaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE ?? "", ...(bootRepoUrl ? { bootRepoUrl } : {}), sourceCommit: options.runnerJobDefaults?.sourceCommit ?? options.sourceCommit, diff --git a/src/mgr/server.ts b/src/mgr/server.ts index c6a0b6e..4cf2652 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -46,10 +46,15 @@ function runnerJobDefaultsForRequest(defaults: ManagerServerOptions["runnerJobDe const serviceAccountName = defaults?.serviceAccountName ?? process.env.AGENTRUN_RUNNER_SERVICE_ACCOUNT ?? "agentrun-v01-runner"; const jobNamePrefix = defaults?.jobNamePrefix ?? process.env.AGENTRUN_RUNNER_JOB_NAME_PREFIX ?? serviceAccountName; const lane = defaults?.lane ?? process.env.AGENTRUN_LANE ?? "v0.1"; + const runnerApiKeySecretRef = defaults?.runnerApiKeySecretRef ?? { + name: process.env.AGENTRUN_RUNNER_API_KEY_SECRET_NAME ?? "", + key: process.env.AGENTRUN_RUNNER_API_KEY_SECRET_KEY ?? "", + }; const retention = runnerRetentionOptionsForRequest(defaults?.retention, namespace, jobNamePrefix, defaults?.kubectlCommand); return { namespace, managerUrl: defaults?.managerUrl ?? process.env.AGENTRUN_INTERNAL_MGR_URL ?? `http://agentrun-mgr.${namespace}.svc.cluster.local:8080`, + runnerApiKeySecretRef, image: defaults?.image ?? process.env.AGENTRUN_RUNNER_IMAGE ?? "", ...optionalStringRecord("bootRepoUrl", defaults?.bootRepoUrl ?? process.env.AGENTRUN_BOOT_REPO_URL), sourceCommit, @@ -76,6 +81,8 @@ function assertRunnerDispatcherRuntimeConfigured(enabled: boolean, defaults: Man ["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_RUNNER_API_KEY_SECRET_NAME", defaults?.runnerApiKeySecretRef?.name ?? process.env.AGENTRUN_RUNNER_API_KEY_SECRET_NAME], + ["AGENTRUN_RUNNER_API_KEY_SECRET_KEY", defaults?.runnerApiKeySecretRef?.key ?? process.env.AGENTRUN_RUNNER_API_KEY_SECRET_KEY], ["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); @@ -162,6 +169,7 @@ export interface ManagerServerOptions { runnerJobDefaults?: { namespace?: string; managerUrl?: string; + runnerApiKeySecretRef?: { name: string; key: string }; image?: string; bootRepoUrl?: string; envIdentity?: string; diff --git a/src/mgr/store.ts b/src/mgr/store.ts index e5311e7..9186d71 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -253,6 +253,12 @@ export class MemoryAgentRunStore implements AgentRunStore { claimRunnerDispatchIntents(input: { owner: string; leaseMs: number; limit: number }): RunnerDispatchIntentRecord[] { const now = Date.now(); + for (const intent of this.runnerDispatchIntents.values()) { + if (!isActiveRunnerDispatchIntent(intent)) continue; + const run = this.getRun(intent.runId); + const command = this.getCommand(intent.commandId); + if (isTerminalRunStatus(run.status) || isTerminalCommandState(command.state)) this.cancelRunnerDispatchIntent(intent, "run or command is terminal"); + } const claimed = Array.from(this.runnerDispatchIntents.values()) .filter((intent) => (intent.state === "pending" || intent.state === "retry" || (intent.state === "dispatching" && Date.parse(intent.leaseExpiresAt ?? "") <= now)) && Date.parse(intent.nextAttemptAt) <= now) .sort((a, b) => a.nextAttemptAt.localeCompare(b.nextAttemptAt) || a.createdAt.localeCompare(b.createdAt)) @@ -266,7 +272,7 @@ export class MemoryAgentRunStore implements AgentRunStore { completeRunnerDispatchIntent(claim: DurableQueueClaim, completion: RunnerDispatchCompletion, eventPayload: JsonRecord): RunnerDispatchIntentRecord { const intent = this.requireRunnerDispatchIntent(claim.id); - assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching"); + this.fenceRunnerDispatchClaim(intent, claim); const event = this.prepareEvent(intent.runId, "backend_status", eventPayload); const at = nowIso(); const updated: RunnerDispatchIntentRecord = { ...intent, state: "dispatched", ...completion, leaseOwner: null, leaseExpiresAt: null, lastError: null, dispatchedAt: at, updatedAt: at }; @@ -277,7 +283,7 @@ export class MemoryAgentRunStore implements AgentRunStore { retryRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord, nextAttemptAt: string, eventPayload: JsonRecord): RunnerDispatchIntentRecord { const intent = this.requireRunnerDispatchIntent(claim.id); - assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching"); + this.fenceRunnerDispatchClaim(intent, claim); 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(intent.id, updated); @@ -287,7 +293,7 @@ export class MemoryAgentRunStore implements AgentRunStore { terminalizeRunnerDispatchIntent(claim: DurableQueueClaim, error: JsonRecord): JsonRecord { const intent = this.requireRunnerDispatchIntent(claim.id); - assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching"); + this.fenceRunnerDispatchClaim(intent, claim); const at = nowIso(); const failure = redactJson(error); const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed"; @@ -299,6 +305,7 @@ export class MemoryAgentRunStore implements AgentRunStore { const run = this.getRun(intent.runId); const updatedRun = isTerminalRunStatus(run.status) ? run : this.updateRun(run.id, { status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: message }); if (!isTerminalRunStatus(run.status)) { + this.cancelRunnerDispatchIntentsForRun(run.id, "run terminalized by runner dispatch failure", at); this.appendEvent(run.id, "backend_status", { phase: "runner-dispatch-failed", commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); this.appendEvent(run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message, threadId: null, turnId: null }); this.appendEvent(run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message }); @@ -438,11 +445,13 @@ export class MemoryAgentRunStore implements AgentRunStore { finishCommand(commandId: string, result: Pick): CommandRecord { const command = this.getCommand(commandId); if (isTerminalCommandState(command.state)) { + this.cancelRunnerDispatchIntentForCommand(commandId, "command is terminal"); if (command.state === "cancelled" && result.terminalStatus !== "cancelled") this.appendLateWriteRejected(command.runId, command, result, "command-status"); return command; } const next = { ...command, state: commandStateFromTerminal(result.terminalStatus), updatedAt: nowIso() }; this.commands.set(commandId, next); + this.cancelRunnerDispatchIntentForCommand(commandId, `command terminalized as ${next.state}`, next.updatedAt); const run = this.getRun(command.runId); if (result.threadId && run.sessionRef?.sessionId) this.upsertSessionThread(run, result.threadId, result.turnId ?? null); if (command.type === "turn") this.touchSessionForRun(this.getRun(command.runId), { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: next.updatedAt }, { bumpVersion: true, at: next.updatedAt }); @@ -489,11 +498,13 @@ export class MemoryAgentRunStore implements AgentRunStore { finishRun(runId: string, result: Pick): RunRecord { const existing = this.getRun(runId); if (isTerminalRunStatus(existing.status)) { + this.cancelRunnerDispatchIntentsForRun(runId, "run is terminal"); if (existing.status === "cancelled" && result.terminalStatus !== "cancelled") this.appendLateWriteRejected(runId, null, result, "run-status"); return existing; } const status = statusFromTerminal(result.terminalStatus); const next = this.updateRun(runId, { status, terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: result.failureMessage }); + this.cancelRunnerDispatchIntentsForRun(runId, `run terminalized as ${status}`, next.updatedAt); if (result.threadId && next.sessionRef?.sessionId) this.upsertSessionThread(next, result.threadId, result.turnId ?? null); this.appendEvent(runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage }); this.touchSessionForRun(this.getRun(runId), { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: next.updatedAt }, { bumpVersion: true, at: next.updatedAt }); @@ -502,7 +513,10 @@ export class MemoryAgentRunStore implements AgentRunStore { cancelRun(runId: string, reason = "cancel requested"): RunRecord { const run = this.getRun(runId); - if (isTerminalRunStatus(run.status)) return run; + if (isTerminalRunStatus(run.status)) { + this.cancelRunnerDispatchIntentsForRun(runId, reason); + return run; + } const at = nowIso(); const cancel = this.createCancelRequest({ targetKind: "run", targetId: runId, run, command: null, reason, at, stage: "accepted" }); this.appendCancelStage(runId, "accepted", cancel); @@ -519,6 +533,7 @@ export class MemoryAgentRunStore implements AgentRunStore { this.commands.set(command.id, cancelled); this.appendEvent(runId, "backend_status", { phase: "command-terminal", commandId: command.id, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", cancelRequestId: cancel.id, cancelEpoch: cancel.epoch }); } + this.cancelRunnerDispatchIntentsForRun(runId, reason, at); const next = this.updateRun(runId, { status: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: reason }); this.appendEvent(runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason, cancelRequestId: cancel.id, cancelEpoch: cancel.epoch }); this.appendCancelStage(runId, "terminalized", cancel); @@ -528,13 +543,17 @@ export class MemoryAgentRunStore implements AgentRunStore { cancelCommand(commandId: string, reason = "cancel requested"): CommandRecord { const command = this.getCommand(commandId); - if (isTerminalCommandState(command.state)) return command; + if (isTerminalCommandState(command.state)) { + this.cancelRunnerDispatchIntentForCommand(commandId, reason); + return command; + } const run = this.getRun(command.runId); const at = nowIso(); const cancel = this.createCancelRequest({ targetKind: "command", targetId: commandId, run, command, reason, at, stage: "accepted" }); this.appendCancelStage(command.runId, "accepted", cancel); const next = { ...command, state: "cancelled" as const, cancelEpoch: cancel.epoch, cancelRequestId: cancel.id, cancelRequestedAt: at, cancelReason: reason, updatedAt: at }; this.commands.set(commandId, next); + this.cancelRunnerDispatchIntentForCommand(commandId, reason, at); this.appendCancelStage(command.runId, "persisted", cancel); const leaseExpired = Boolean(run.claimedBy && isLeaseExpired(run.leaseExpiresAt)); if (leaseExpired) this.appendCancelStage(command.runId, "fenced", cancel, { claimedBy: run.claimedBy, leaseExpiresAt: run.leaseExpiresAt }); @@ -787,6 +806,29 @@ export class MemoryAgentRunStore implements AgentRunStore { return intent; } + private fenceRunnerDispatchClaim(intent: RunnerDispatchIntentRecord, claim: DurableQueueClaim): void { + assertMemoryClaim(intent, claim, "runner dispatch", intent.state === "dispatching"); + const run = this.getRun(intent.runId); + const command = this.getCommand(intent.commandId); + if (!isTerminalRunStatus(run.status) && !isTerminalCommandState(command.state)) return; + this.cancelRunnerDispatchIntent(intent, "run or command became terminal during dispatch"); + throw staleClaimError("runner dispatch", claim); + } + + private cancelRunnerDispatchIntentForCommand(commandId: string, reason: string, at = nowIso()): void { + const intent = this.getRunnerDispatchIntent(commandId); + if (intent) this.cancelRunnerDispatchIntent(intent, reason, at); + } + + private cancelRunnerDispatchIntentsForRun(runId: string, reason: string, at = nowIso()): void { + for (const intent of this.runnerDispatchIntents.values()) if (intent.runId === runId) this.cancelRunnerDispatchIntent(intent, reason, at); + } + + private cancelRunnerDispatchIntent(intent: RunnerDispatchIntentRecord, reason: string, at = nowIso()): void { + if (!isActiveRunnerDispatchIntent(intent)) return; + this.runnerDispatchIntents.set(intent.id, { ...intent, state: "cancelled", leaseOwner: null, leaseExpiresAt: null, lastError: { code: "cancelled", message: reason, valuesPrinted: false }, updatedAt: at }); + } + private requireKafkaEventOutbox(outboxId: string): KafkaEventOutboxRecord { const item = this.kafkaEventOutbox.get(outboxId); if (!item) throw new AgentRunError("schema-invalid", `Kafka event outbox ${outboxId} was not found`, { httpStatus: 404 }); @@ -1189,10 +1231,18 @@ function commandForEvent(commands: Map, event: RunEvent): 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 } }); + throw staleClaimError(component, claim); } } +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 isActiveRunnerDispatchIntent(intent: RunnerDispatchIntentRecord): boolean { + return intent.state === "pending" || intent.state === "dispatching" || intent.state === "retry"; +} + 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/runner/k8s-job.ts b/src/runner/k8s-job.ts index ff0b5b3..418261a 100644 --- a/src/runner/k8s-job.ts +++ b/src/runner/k8s-job.ts @@ -45,6 +45,7 @@ export interface RunnerJobRenderOptions { commandId: string; runnerJobId?: string; managerUrl: string; + runnerApiKeySecretRef: RunnerApiKeySecretRef; image: string; bootRepoUrl?: string; namespace?: string; @@ -66,6 +67,11 @@ export interface RunnerJobRenderOptions { dryRun?: boolean; } +export interface RunnerApiKeySecretRef { + name: string; + key: string; +} + export interface RunnerSessionPvcOptions { pvcName: string; namespace: string; @@ -136,6 +142,7 @@ export function renderRunnerJobDryRun(options: RunnerJobRenderOptions): JsonReco managerUrl: options.managerUrl, sourceCommit: render.sourceCommit, }, + runnerApiKeySecretRef: { ...options.runnerApiKeySecretRef, valuesPrinted: false }, secretRefs: render.secretRefs.map((item) => ({ profile: item.profile, name: item.secretRef.name, namespace: item.secretRef.namespace ?? render.namespace, keys: item.secretRef.keys ?? [], mountPath: item.runtimeMountPath, projectionPath: item.projectionMountPath, writableCopy: true, valuesPrinted: false })), toolCredentials: summarizeToolCredentials(render.toolCredentials, render.namespace), gitTransport: gitTransportSummary(), @@ -164,6 +171,7 @@ export function renderRunnerJobManifest(options: RunnerJobRenderOptions): { mani const serviceAccountName = options.serviceAccountName ?? "agentrun-v01-runner"; const jobNamePrefix = normalizeJobNamePrefix(options.jobNamePrefix ?? serviceAccountName); const lane = options.lane ?? process.env.AGENTRUN_LANE ?? "v0.1"; + const runnerApiKeySecretRef = normalizeRunnerApiKeySecretRef(options.runnerApiKeySecretRef); const warnings: string[] = []; const ttlSecondsAfterFinished = normalizeTtlSecondsAfterFinished(options.ttlSecondsAfterFinished, warnings); const ttlPolicy = terminalArtifactTtlPolicy(ttlSecondsAfterFinished); @@ -176,7 +184,7 @@ export function renderRunnerJobManifest(options: RunnerJobRenderOptions): { mani const toolCredentials = toolCredentialProjections(options.run, namespace); const sessionPvc = options.sessionPvc; if (secretRefs.length === 0) warnings.push("run executionPolicy.secretScope 未声明 provider SecretRef；runner 将按 secret-unavailable 上报，而不会降级直连外部凭据"); - const env = runnerEnv(options, { namespace, jobName, runnerJobId, runnerId, attemptId, sourceCommit, secretRefs, toolCredentials, sessionPvc, runnerIdleTimeoutMs, backendRetryMaxAttempts, backendRetryInitialBackoffMs, backendRetryMaxBackoffMs }); + const env = runnerEnv(options, { namespace, jobName, runnerJobId, runnerId, attemptId, sourceCommit, runnerApiKeySecretRef, secretRefs, toolCredentials, sessionPvc, runnerIdleTimeoutMs, backendRetryMaxAttempts, backendRetryInitialBackoffMs, backendRetryMaxBackoffMs }); const manifest: JsonRecord = { apiVersion: "batch/v1", kind: "Job", @@ -252,13 +260,13 @@ export function renderRunnerJobManifest(options: RunnerJobRenderOptions): { mani return { manifest, namespace, jobName, runnerJobId, runnerId, attemptId, sourceCommit, serviceAccountName, secretRefs, toolCredentials, warnings, ttlSecondsAfterFinished, ttlPolicy, runnerIdleTimeoutMs, backendRetryMaxAttempts, backendRetryInitialBackoffMs, backendRetryMaxBackoffMs }; } -function runnerEnv(options: RunnerJobRenderOptions, context: { namespace: string; jobName: string; runnerJobId: string; runnerId: string; attemptId: string; sourceCommit: string; secretRefs: CredentialProjection[]; toolCredentials: ToolCredentialProjection[]; sessionPvc: RunnerSessionPvcOptions | undefined; runnerIdleTimeoutMs: number; backendRetryMaxAttempts: number; backendRetryInitialBackoffMs: number; backendRetryMaxBackoffMs: number }): JsonRecord[] { +function runnerEnv(options: RunnerJobRenderOptions, context: { namespace: string; jobName: string; runnerJobId: string; runnerId: string; attemptId: string; sourceCommit: string; runnerApiKeySecretRef: RunnerApiKeySecretRef; secretRefs: CredentialProjection[]; toolCredentials: ToolCredentialProjection[]; sessionPvc: RunnerSessionPvcOptions | undefined; runnerIdleTimeoutMs: number; backendRetryMaxAttempts: number; backendRetryInitialBackoffMs: number; backendRetryMaxBackoffMs: number }): JsonRecord[] { const selectedSecret = context.secretRefs.find((item) => item.profile === options.run.backendProfile); const codexHome = runnerCodexHome(options.run.backendProfile, selectedSecret, context.sessionPvc); const bootRepoUrl = optionalString(options.bootRepoUrl) ?? defaultBootRepoUrl; return dedupeEnvVars([ { name: "AGENTRUN_MGR_URL", value: options.managerUrl }, - { name: "AGENTRUN_API_KEY", valueFrom: { secretKeyRef: { name: "agentrun-v01-api-key", key: "HWLAB_API_KEY" } } }, + { name: "AGENTRUN_API_KEY", valueFrom: { secretKeyRef: context.runnerApiKeySecretRef } }, { name: "AGENTRUN_RUN_ID", value: options.run.id }, { name: "AGENTRUN_COMMAND_ID", value: options.commandId }, { name: "AGENTRUN_RUNNER_JOB_ID", value: context.runnerJobId }, @@ -616,6 +624,13 @@ function normalizeJobNamePrefix(value: string): string { return normalized; } +function normalizeRunnerApiKeySecretRef(value: RunnerApiKeySecretRef): RunnerApiKeySecretRef { + const name = optionalString(value?.name); + const key = optionalString(value?.key); + if (!name || !key) throw new Error("runnerApiKeySecretRef.name and runnerApiKeySecretRef.key are required"); + return { name, key }; +} + function shortDnsHash(...parts: string[]): string { return shortHash(parts.join(":")); } diff --git a/src/selftest/cases/20-runner-k8s-job.ts b/src/selftest/cases/20-runner-k8s-job.ts index bce4788..ecb0d1b 100644 --- a/src/selftest/cases/20-runner-k8s-job.ts +++ b/src/selftest/cases/20-runner-k8s-job.ts @@ -9,6 +9,8 @@ import { renderRunnerJobDryRun } from "../../runner/k8s-job.js"; import type { JsonRecord, RunRecord } from "../../common/types.js"; import { assertNoSecretLeak, createRunWithCommand, loadArtificerImageRef, type SelfTestCase } from "../harness.js"; +const runnerApiKeySecretRef = { name: "agentrun-selftest-api-key", key: "HWLAB_API_KEY" }; + const selfTest: SelfTestCase = async (context) => { const server = await startManagerServer({ port: 0, host: "127.0.0.1", sourceCommit: "self-test", store: new MemoryAgentRunStore() }); try { @@ -38,6 +40,7 @@ const selfTest: SelfTestCase = async (context) => { run: await client.get(`/api/v1/runs/${item.runId}`) as RunRecord, commandId: item.commandId, managerUrl: server.baseUrl, + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", attemptId: "attempt_selftest", sourceCommit: "self-test", @@ -56,6 +59,8 @@ const selfTest: SelfTestCase = async (context) => { assertRunnerJobUsesG14EgressProxy(rendered.manifest as JsonRecord); assertRunnerJobUsesBoundedGitTransport(rendered); assert.equal(runnerEnvValue(rendered.manifest as JsonRecord, "AGENTRUN_CODEX_SHELL_SANDBOX"), "danger-full-access"); + assert.deepEqual(runnerEnvSecretKeyRef(rendered.manifest as JsonRecord, "AGENTRUN_API_KEY"), runnerApiKeySecretRef); + assert.deepEqual(rendered.runnerApiKeySecretRef, { ...runnerApiKeySecretRef, valuesPrinted: false }); assert.equal(runnerEnvValue(rendered.manifest as JsonRecord, "HWLAB_API_KEY"), "REDACTED"); assert.deepEqual((((rendered.transientEnv as JsonRecord).names) as string[]), ["HWLAB_API_KEY"]); assertNoSecretLeak(rendered); @@ -78,6 +83,7 @@ const selfTest: SelfTestCase = async (context) => { run: await client.get(`/api/v1/runs/${deepseekItem.runId}`) as RunRecord, commandId: deepseekItem.commandId, managerUrl: server.baseUrl, + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", attemptId: "attempt_selftest_deepseek", sourceCommit: "self-test", @@ -91,6 +97,7 @@ const selfTest: SelfTestCase = async (context) => { run: await client.get(`/api/v1/runs/${minimaxItem.runId}`) as RunRecord, commandId: minimaxItem.commandId, managerUrl: server.baseUrl, + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", attemptId: "attempt_selftest_minimax_m3", sourceCommit: "self-test", @@ -105,6 +112,7 @@ const selfTest: SelfTestCase = async (context) => { run: await client.get(`/api/v1/runs/${dsflashGoItem.runId}`) as RunRecord, commandId: dsflashGoItem.commandId, managerUrl: server.baseUrl, + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", attemptId: "attempt_selftest_dsflash_go", sourceCommit: "self-test", @@ -180,6 +188,7 @@ process.exit(1); runnerJobDefaults: { namespace: "agentrun-v01", managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080", + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", envIdentity: "selftest-env-identity", kubectlCommand: fakeKubectl, @@ -338,6 +347,7 @@ process.exit(1); runnerJobDefaults: { namespace: "agentrun-v02", managerUrl: "http://agentrun-mgr.agentrun-v02.svc.cluster.local:8080", + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", serviceAccountName: "agentrun-v02-runner", jobNamePrefix: "agentrun-v02-runner", @@ -463,6 +473,7 @@ process.exit(1); run: sessionRunRecord, commandId: "cmd-selftest-session-pvc", managerUrl: server.baseUrl, + runnerApiKeySecretRef, image: "127.0.0.1:5000/agentrun/agentrun-mgr-env:self-test", sessionPvc: { pvcName: "agentrun-v01-session-sess-selftest-runner-001", namespace: "agentrun-v01", mountPath: "/home/agentrun/.agentrun-sessions/sessions", codexRolloutSubdir: "sessions" }, }); @@ -503,6 +514,10 @@ function runnerEnvValue(manifest: JsonRecord, name: string): unknown { return runnerEnvEntry(manifest, name)?.value; } +function runnerEnvSecretKeyRef(manifest: JsonRecord, name: string): JsonRecord | undefined { + return (runnerEnvEntry(manifest, name)?.valueFrom as JsonRecord | undefined)?.secretKeyRef as JsonRecord | undefined; +} + function assertRunnerJobUsesTransientEnvSecret(manifest: JsonRecord, envName: string, secretName: string): void { const entry = runnerEnvEntry(manifest, envName); assert.ok(entry, `${envName} env should be present`); diff --git a/src/selftest/cases/21-runner-durable-dispatch.ts b/src/selftest/cases/21-runner-durable-dispatch.ts index ec9a8e3..62fed5d 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 { access, chmod, rm } from "node:fs/promises"; import path from "node:path"; -import type { CreateRunInput, RunEvent } from "../../common/types.js"; +import type { CreateRunInput, RunEvent, RunRecord } 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"; @@ -47,6 +47,8 @@ const selfTest: SelfTestCase = async (context) => { await assertClaimsAreLeasedJustInTime(fakeKubectl); await assertSettleEventFailureDoesNotHalfCommit(fakeKubectl); await assertEnabledDispatcherRejectsLegacyDefaults(); + assertTerminalTransitionsFenceDispatchIntents(); + assertSettleFencesNewlyTerminalParent(); process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "fail"; const failedRun = store.createRun({ ...runInput(), sessionRef: null }); @@ -65,7 +67,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", "active-runner-reuse", "attempt-timeout-recovery", "just-in-time-claim", "atomic-settle-event", "enabled-config-fail-fast", "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", "terminal-intent-fencing", "settle-secondary-fence", "permanent-failure-terminal"] }; }; async function assertHttpAdmissionContract(kubectlCommand: string): Promise { @@ -103,6 +105,7 @@ function defaults(kubectlCommand: string): RunnerJobDefaults { return { namespace: "agentrun-v02", managerUrl: "http://agentrun-mgr.agentrun-v02.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-v02-api-key", key: "HWLAB_API_KEY" }, image, sourceCommit: "selftest", serviceAccountName: "agentrun-v02-runner", @@ -190,7 +193,7 @@ class FailingDispatchEventStore extends MemoryAgentRunStore { } 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 keys = ["AGENTRUN_RUNTIME_NAMESPACE", "AGENTRUN_INTERNAL_MGR_URL", "AGENTRUN_RUNNER_IMAGE", "AGENTRUN_RUNNER_SERVICE_ACCOUNT", "AGENTRUN_RUNNER_JOB_NAME_PREFIX", "AGENTRUN_RUNNER_API_KEY_SECRET_NAME", "AGENTRUN_RUNNER_API_KEY_SECRET_KEY", "AGENTRUN_LANE"] as const; const previous = Object.fromEntries(keys.map((key) => [key, process.env[key]])); keys.forEach((key) => { delete process.env[key]; }); try { @@ -207,6 +210,60 @@ async function assertEnabledDispatcherRejectsLegacyDefaults(): Promise { ); } +function assertTerminalTransitionsFenceDispatchIntents(): void { + const cancelledRunStore = new MemoryAgentRunStore(); + const cancelledRun = cancelledRunStore.createRun(runInput()); + const cancelledRunCommand = cancelledRunStore.createCommand(cancelledRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } })); + cancelledRunStore.cancelRun(cancelledRun.id, "selftest cancel run"); + assertCancelledIntent(cancelledRunStore, cancelledRunCommand.id); + assert.equal(cancelledRunStore.claimRunnerDispatchIntents({ owner: "late-dispatcher", leaseMs: 10_000, limit: 1 }).length, 0); + + const cancelledCommandStore = new MemoryAgentRunStore(); + const cancelledCommandRun = cancelledCommandStore.createRun(runInput()); + const cancelledCommand = cancelledCommandStore.createCommand(cancelledCommandRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } })); + const claim = cancelledCommandStore.claimRunnerDispatchIntents({ owner: "inflight-dispatcher", leaseMs: 10_000, limit: 1 })[0]; + assert.ok(claim); + cancelledCommandStore.cancelCommand(cancelledCommand.id, "selftest cancel command"); + assertCancelledIntent(cancelledCommandStore, cancelledCommand.id); + assert.throws(() => cancelledCommandStore.completeRunnerDispatchIntent(claim, { dispatchOutcome: "kubernetes-job", actualRunnerJobId: "job-too-late", activeRunnerId: null }, {}), /claim is stale/u); + + const finishedStore = new MemoryAgentRunStore(); + const finishedRun = finishedStore.createRun(runInput()); + const finishedCommand = finishedStore.createCommand(finishedRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } })); + finishedStore.finishCommand(finishedCommand.id, { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: null, turnId: null }); + assertCancelledIntent(finishedStore, finishedCommand.id); + const runTerminalCommand = finishedStore.createCommand(finishedRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } })); + finishedStore.finishRun(finishedRun.id, { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: null, turnId: null }); + assertCancelledIntent(finishedStore, runTerminalCommand.id); +} + +function assertSettleFencesNewlyTerminalParent(): void { + const store = new TerminalProjectionStore(); + const run = store.createRun(runInput()); + const command = store.createCommand(run.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: { image } } })); + const claim = store.claimRunnerDispatchIntents({ owner: "race-dispatcher", leaseMs: 10_000, limit: 1 })[0]; + assert.ok(claim); + store.projectTerminal = true; + assert.throws(() => store.retryRunnerDispatchIntent(claim, { message: "too late" }, new Date().toISOString(), {}), /claim is stale/u); + assertCancelledIntent(store, command.id); +} + +function assertCancelledIntent(store: MemoryAgentRunStore, commandId: string): void { + const intent = store.getRunnerDispatchIntent(commandId); + assert.equal(intent?.state, "cancelled"); + assert.equal(intent?.leaseOwner, null); + assert.equal(intent?.leaseExpiresAt, null); +} + +class TerminalProjectionStore extends MemoryAgentRunStore { + projectTerminal = false; + + override getRun(runId: string): RunRecord { + const run = super.getRun(runId); + return this.projectTerminal ? { ...run, status: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" } : run; + } +} + async function waitFor(predicate: () => boolean, timeoutMs: number, message: string | (() => string)): Promise { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { diff --git a/src/selftest/cases/45-provider-profile-management.ts b/src/selftest/cases/45-provider-profile-management.ts index 3398def..b8b3b7a 100644 --- a/src/selftest/cases/45-provider-profile-management.ts +++ b/src/selftest/cases/45-provider-profile-management.ts @@ -146,6 +146,7 @@ process.exit(1); runnerJobDefaults: { namespace: "agentrun-v01", managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-selftest-api-key", key: "HWLAB_API_KEY" }, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:2222222222222222222222222222222222222222222222222222222222222222", kubectlCommand: fakeKubectl, }, diff --git a/src/selftest/cases/50-hwlab-manual-dispatch.ts b/src/selftest/cases/50-hwlab-manual-dispatch.ts index 7c5a2dc..f4c2d2c 100644 --- a/src/selftest/cases/50-hwlab-manual-dispatch.ts +++ b/src/selftest/cases/50-hwlab-manual-dispatch.ts @@ -56,6 +56,7 @@ process.exit(1); runnerJobDefaults: { namespace: "agentrun-v01", managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-selftest-api-key", key: "HWLAB_API_KEY" }, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", kubectlCommand: fakeKubectl, }, diff --git a/src/selftest/cases/60-hwlab-baseline-contract.ts b/src/selftest/cases/60-hwlab-baseline-contract.ts index 36d95f8..d76b74f 100644 --- a/src/selftest/cases/60-hwlab-baseline-contract.ts +++ b/src/selftest/cases/60-hwlab-baseline-contract.ts @@ -35,6 +35,7 @@ console.log(JSON.stringify({ apiVersion: manifest.apiVersion, kind: manifest.kin runnerJobDefaults: { namespace: "agentrun-v01", managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-selftest-api-key", key: "HWLAB_API_KEY" }, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", kubectlCommand: fakeKubectl, }, diff --git a/src/selftest/cases/75-queue-q2-dispatch.ts b/src/selftest/cases/75-queue-q2-dispatch.ts index fa16653..af94eb7 100644 --- a/src/selftest/cases/75-queue-q2-dispatch.ts +++ b/src/selftest/cases/75-queue-q2-dispatch.ts @@ -58,6 +58,7 @@ process.exit(1); runnerJobDefaults: { namespace: "agentrun-v01", managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-selftest-api-key", key: "HWLAB_API_KEY" }, image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", envIdentity: "selftest-env-identity", kubectlCommand: fakeKubectl,