修复 durable dispatch 终态隔离与 runner 凭据引用

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(),

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<LockedRunnerDispatchClaim> {
+  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<number> {
+  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<number> {
+  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"]);

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<RunnerDispatchIntentRecord[]> {
     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<RunnerDispatchIntentRecord> {
-    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<RunnerDispatchIntentRecord> {
-    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<JsonRecord> {
-    return this.withTransaction(async (client) => {
-      const intentResult = await client.query(
-        "SELECT * FROM agentrun_runner_dispatch_intents WHERE id = $1 AND state = 'dispatching' AND lease_owner = $2 AND attempt_count = $3 AND lease_expires_at > now() FOR UPDATE",
-        [claim.id, claim.leaseOwner, claim.attemptCount],
-      );
-      if (!intentResult.rows[0]) throw staleClaimError("runner dispatch", claim);
-      const intent = runnerDispatchIntentFromRow(intentResult.rows[0]);
-      const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [intent.commandId]);
-      if (!commandResult.rows[0]) throw new AgentRunError("schema-invalid", `command ${intent.commandId} was not found`, { httpStatus: 404 });
-      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<JsonRecord> {
@@ -924,20 +929,24 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
 
   async finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): Promise<CommandRecord> {
     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<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): Promise<RunRecord> {
     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<RunRecord> {
     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<CommandRecord> {
     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),

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,

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;

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<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): 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<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): 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<string, CommandRecord>, 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<T extends JsonRecord>(items: T[], component: string, completed: (item: T) => boolean = (item) => ["dispatched", "failed", "cancelled"].includes(String(item.state ?? ""))): JsonRecord {
   const pending = items.filter((item) => !completed(item));
   const retrying = pending.filter((item) => Number(item.attemptCount ?? 0) > 0);

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(":"));
 }

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`);

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<void> {
@@ -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<void> {
-  const keys = ["AGENTRUN_RUNTIME_NAMESPACE", "AGENTRUN_INTERNAL_MGR_URL", "AGENTRUN_RUNNER_IMAGE", "AGENTRUN_RUNNER_SERVICE_ACCOUNT", "AGENTRUN_RUNNER_JOB_NAME_PREFIX", "AGENTRUN_LANE"] as const;
+  const 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<void> {
   );
 }
 
+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<void> {
   const deadline = Date.now() + timeoutMs;
   while (Date.now() < deadline) {

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,
     },

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,
     },

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,
     },

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,