fix: 用 DB 与 K8s CAS 关闭 retention 竞态

src/mgr/kubernetes-runner-job.ts

@@ -79,6 +79,12 @@ export interface CreateRunnerJobInput extends JsonRecord {
 }
 
 export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise<JsonRecord> {
+  if (!options.defaults.retention) return await createKubernetesRunnerJobUnderFence(options);
+  const namespace = optionalString(options.input.namespace) ?? options.defaults.namespace;
+  return await options.store.withRunnerCreateFence(namespace, async () => await createKubernetesRunnerJobUnderFence(options));
+}
+
+async function createKubernetesRunnerJobUnderFence(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise<JsonRecord> {
   throwIfAborted(options.signal, "runner dispatch attempt");
   const commandId = stringField(options.input, "commandId");
   const run = await options.store.getRun(options.runId);

src/mgr/postgres-store.ts

@@ -5,7 +5,7 @@ import { AgentRunError } from "../common/errors.js";
 import { redactJson } from "../common/redaction.js";
 import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
 import { newId, nowIso, stableHash } from "../common/validation.js";
-import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
+import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
 import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, lateWriteRejectedPayload, parseQueueCursor, parseSessionCursor, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
 import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
 import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
@@ -458,12 +458,14 @@ export async function createPostgresAgentRunStore(options: PostgresStoreOptions)
 
 export class PostgresAgentRunStore implements AgentRunStore {
   private readonly pool: Pool;
+  private readonly runnerCreateFencePool: Pool;
   private readonly eventOutboxConfig: EventOutboxConfig;
   private migrationReady = false;
   private appliedMigrationId: string | null = null;
 
   constructor(options: PostgresStoreOptions) {
     this.pool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-v01", ...(options.poolMax === undefined ? {} : { max: options.poolMax }) });
+    this.runnerCreateFencePool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-runner-capacity-fence", max: 1 });
     this.eventOutboxConfig = options.eventOutbox ?? { enabled: false, topic: null, source: null };
   }
 
@@ -879,6 +881,50 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     });
   }
 
+  async withRunnerCreateFence<T>(namespace: string, operation: () => Promise<T>): Promise<T> {
+    const client = await this.runnerCreateFencePool.connect();
+    try {
+      await client.query("BEGIN");
+      await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-runner-create", namespace]);
+      const result = await operation();
+      await client.query("COMMIT");
+      return result;
+    } catch (error) {
+      await client.query("ROLLBACK");
+      throw error;
+    } finally {
+      client.release();
+    }
+  }
+
+  async withRunnerRetentionFence<T>(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise<T>): Promise<T> {
+    return this.withTransaction(async (client) => {
+      const run = await this.requireRunForUpdate(client, input.runId);
+      const commandRows = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [run.id]);
+      const commands = commandRows.rows.slice(0, 100).map(commandFromRow);
+      const commandRow = commandRows.rows.find((item) => String(item.id) === input.commandId);
+      if (!commandRow) throw new AgentRunError("schema-invalid", `command ${input.commandId} does not belong to run ${input.runId}`, { httpStatus: 409 });
+      const command = commandFromRow(commandRow);
+      const eventRows = await client.query("SELECT * FROM agentrun_events WHERE run_id = $1 ORDER BY seq ASC LIMIT 500", [run.id]);
+      const result = await operation({ run, command, commands, events: eventRows.rows.map(eventFromRow) });
+      if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) return result;
+
+      const at = nowIso();
+      const updatedCommands = await client.query("UPDATE agentrun_commands SET state = 'failed', updated_at = $2 WHERE run_id = $1 AND state NOT IN ('completed', 'failed', 'cancelled') RETURNING *", [run.id, at]);
+      await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, input.reason, at);
+      const updatedRun = await client.query("UPDATE agentrun_runs SET status = 'failed', terminal_status = 'failed', failure_kind = 'infra-failed', failure_message = $2, claimed_by = NULL, lease_expires_at = NULL, updated_at = $3 WHERE id = $1 RETURNING *", [run.id, input.reason, at]);
+      await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [run.id]);
+      for (const row of updatedCommands.rows) {
+        const item = commandFromRow(row);
+        await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", commandId: item.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message: input.reason, retentionFence: true });
+      }
+      await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message: input.reason, retentionFence: true });
+      const next = runFromRow(updatedRun.rows[0]);
+      await this.touchSessionForRun(client, next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at });
+      return result;
+    });
+  }
+
   async claimRun(runId: string, runnerId: string, leaseMs: number): Promise<RunRecord> {
     return this.withTransaction(async (client) => {
       const run = await this.requireRunForUpdate(client, runId);
@@ -1393,7 +1439,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
   }
 
   async close(): Promise<void> {
-    await this.pool.end();
+    await Promise.all([this.pool.end(), this.runnerCreateFencePool.end()]);
   }
 
   private async createCancelRequest(client: PoolClient, input: { targetKind: CancelTargetKind; targetId: string; run: RunRecord; command: CommandRecord | null; reason: string; at: string; stage: CancelStage }): Promise<CancelRequestRecord> {

src/mgr/runner-retention.ts

@@ -166,12 +166,26 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
   let deletedAssociatedResourceCount = 0;
   for (const item of selected) {
     await revalidateSelectedRunner(input.store, input.options, item);
+    if (!item.runId || !item.commandId || !item.dbFactsFingerprint) {
+      throw new AgentRunError("infra-failed", "runner retention selected resource has no fenceable database identity", {
+        httpStatus: 409,
+        details: { reason: "runner-retention-db-fence-unavailable", namespace: input.options.namespace, resourceKind: item.resourceKind, name: item.name, valuesPrinted: false },
+      });
+    }
+    await input.store.withRunnerRetentionFence({
+      runId: item.runId,
+      commandId: item.commandId,
+      terminalizeStalePending: item.candidateKind === "stale-pending",
+      reason: `runner retention retired stale Kubernetes ${item.resourceKind}/${item.name}`,
+    }, async (facts) => {
+      const currentDbFingerprint = databaseFactsFingerprint(facts.run, facts.command, facts.commands, facts.events);
+      if (currentDbFingerprint !== item.dbFactsFingerprint) throw databaseFenceRejected(input.options, item, currentDbFingerprint, facts.run);
+      await deleteRunnerResourceCas(input.options, item);
+    });
     if (item.resourceKind === "job") {
-      await deleteRunnerResource(input.options, item);
       deletedRunnerJobCount++;
       deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
     } else {
-      await deleteRunnerResource(input.options, item);
       deletedRunnerPodCount++;
     }
   }
@@ -430,16 +444,74 @@ async function revalidateSelectedRunner(store: AgentRunStore, options: RunnerRet
   });
 }
 
-async function deleteRunnerResource(options: RunnerRetentionOptions, item: RunnerResourceEntry): Promise<void> {
-  const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", item.resourceKind, item.name, "-n", options.namespace], options.signal);
-  if (result.code === 0) return;
-  throw new AgentRunError("infra-failed", `kubectl delete ${item.resourceKind} failed with code ${result.code}`, {
-    httpStatus: 502,
-    details: redactJson({
-      reason: "runner-retention-delete-failed",
+function databaseFenceRejected(options: RunnerRetentionOptions, item: RunnerResourceEntry, currentDbFingerprint: string, run: RunRecord): AgentRunError {
+  return new AgentRunError("infra-failed", "runner retention database facts changed before atomic deletion", {
+    httpStatus: 409,
+    details: {
+      reason: "runner-retention-db-fence-rejected",
+      changeReason: run.claimedBy || run.status === "claimed" || run.status === "running" ? "run-became-active" : "database-fingerprint-changed",
       namespace: options.namespace,
       resourceKind: item.resourceKind,
       name: item.name,
+      selectedDbFingerprint: item.dbFactsFingerprint,
+      currentDbFingerprint,
+      runStatus: run.status,
+      claimedBy: run.claimedBy,
+      leaseExpiresAt: run.leaseExpiresAt,
+      valuesPrinted: false,
+    },
+  });
+}
+
+async function deleteRunnerResourceCas(options: RunnerRetentionOptions, item: RunnerResourceEntry): Promise<void> {
+  if (item.resourceKind === "job") {
+    for (const pod of [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name))) {
+      await deleteKubernetesObjectCas(options, item, "pod", pod.name, pod.uid, pod.resourceVersion);
+    }
+    const currentJob = await kubectlGetObject(options, "job", item.name);
+    const expectedIdentity = {
+      "agentrun.pikastech.local/run-id": item.runId ?? "",
+      "agentrun.pikastech.local/command-id": item.commandId ?? "",
+      "agentrun.pikastech.local/runner-id": item.runnerId ?? "",
+    };
+    if (currentJob.metadata?.uid !== item.resourceUid || !resourceIdentityMatches(expectedIdentity, currentJob.metadata?.annotations ?? {})) {
+      throw new AgentRunError("infra-failed", "runner retention Job identity changed after Pod CAS deletion", {
+        httpStatus: 409,
+        details: { reason: "runner-retention-k8s-cas-rejected", changeReason: "job-identity-changed", namespace: options.namespace, resourceKind: "job", name: item.name, expectedUid: item.resourceUid, currentUid: currentJob.metadata?.uid ?? null, valuesPrinted: false },
+      });
+    }
+    await deleteKubernetesObjectCas(options, item, "job", item.name, currentJob.metadata?.uid ?? null, currentJob.metadata?.resourceVersion ?? null);
+    return;
+  }
+  await deleteKubernetesObjectCas(options, item, item.resourceKind, item.name, item.resourceUid, item.resourceVersion);
+}
+
+async function deleteKubernetesObjectCas(options: RunnerRetentionOptions, selected: RunnerResourceEntry, resourceKind: "job" | "pod", name: string, uid: string | null, resourceVersion: string | null): Promise<void> {
+  if (!uid || !resourceVersion) {
+    throw new AgentRunError("infra-failed", "runner retention cannot bind Kubernetes delete preconditions", {
+      httpStatus: 409,
+      details: { reason: "runner-retention-k8s-cas-unavailable", namespace: options.namespace, selectedResourceKind: selected.resourceKind, selectedName: selected.name, resourceKind, name, uidPresent: Boolean(uid), resourceVersionPresent: Boolean(resourceVersion), valuesPrinted: false },
+    });
+  }
+  const resourcePath = resourceKind === "job"
+    ? `/apis/batch/v1/namespaces/${encodeURIComponent(options.namespace)}/jobs/${encodeURIComponent(name)}`
+    : `/api/v1/namespaces/${encodeURIComponent(options.namespace)}/pods/${encodeURIComponent(name)}`;
+  const deleteOptions = JSON.stringify({ apiVersion: "v1", kind: "DeleteOptions", propagationPolicy: "Background", preconditions: { uid, resourceVersion } });
+  const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", "--raw", resourcePath, "-f", "-"], options.signal, deleteOptions);
+  if (result.code === 0) return;
+  const conflict = /\b(?:Conflict|PreconditionFailed|resourceVersion|UID)\b/iu.test(`${result.stderr}\n${result.stdout}`);
+  throw new AgentRunError("infra-failed", `Kubernetes CAS delete ${resourceKind} failed with code ${result.code}`, {
+    httpStatus: conflict ? 409 : 502,
+    details: redactJson({
+      reason: "runner-retention-k8s-cas-rejected",
+      changeReason: conflict ? "object-identity-or-version-changed" : "delete-request-failed",
+      namespace: options.namespace,
+      selectedResourceKind: selected.resourceKind,
+      selectedName: selected.name,
+      resourceKind,
+      name,
+      expectedUid: uid,
+      expectedResourceVersion: resourceVersion,
       exitCode: result.code,
       signal: result.signal,
       stderr: redactText(result.stderr.slice(-2000)),
@@ -460,6 +532,12 @@ async function kubectlGetList(options: RunnerRetentionOptions, resource: string,
   return Array.isArray(items) ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) : [];
 }
 
+async function kubectlGetObject(options: RunnerRetentionOptions, resource: "job" | "pod", name: string): Promise<K8sObject> {
+  const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["get", resource, name, "-n", options.namespace, "-o", "json"], options.signal);
+  if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${resource}/${name} failed with code ${result.code}`, { httpStatus: 409, details: redactJson({ reason: "runner-retention-k8s-cas-rejected", changeReason: "resource-missing-before-cas", namespace: options.namespace, resourceKind: resource, name, stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) });
+  return parseJsonObject(result.stdout, `kubectl get ${resource}/${name}`) as unknown as K8sObject;
+}
+
 async function kubectlGetPodEvents(options: RunnerRetentionOptions): Promise<K8sObject[]> {
   const args = ["get", "events", "-n", options.namespace, "--field-selector", "involvedObject.kind=Pod,type=Warning", "-o", "json"];
   const result = await kubectlRun(options.kubectlCommand ?? "kubectl", args, options.signal);
@@ -480,15 +558,19 @@ async function deleteAssociatedResources(options: RunnerRetentionOptions, jobNam
   return deleted;
 }
 
-async function kubectlRun(kubectlCommand: string, args: string[], abortSignal?: AbortSignal): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
+async function kubectlRun(kubectlCommand: string, args: string[], abortSignal?: AbortSignal, stdin?: string): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
   throwIfAborted(abortSignal, "kubectl");
-  const child = spawn(kubectlCommand, args, { stdio: ["ignore", "pipe", "pipe"] });
+  const child = spawn(kubectlCommand, args, { stdio: [stdin === undefined ? "ignore" : "pipe", "pipe", "pipe"] });
   let stdout = "";
   let stderr = "";
-  child.stdout.setEncoding("utf8");
-  child.stderr.setEncoding("utf8");
-  child.stdout.on("data", (chunk) => { stdout += String(chunk); });
-  child.stderr.on("data", (chunk) => { stderr += String(chunk); });
+  const stdoutStream = child.stdout;
+  const stderrStream = child.stderr;
+  if (!stdoutStream || !stderrStream) throw new AgentRunError("infra-failed", "kubectl output pipes were not created", { httpStatus: 503 });
+  stdoutStream.setEncoding("utf8");
+  stderrStream.setEncoding("utf8");
+  stdoutStream.on("data", (chunk) => { stdout += String(chunk); });
+  stderrStream.on("data", (chunk) => { stderr += String(chunk); });
+  if (stdin !== undefined) child.stdin?.end(stdin);
   const result = await waitForChildProcess(child, { ...(abortSignal ? { signal: abortSignal } : {}), label: "kubectl" });
   return { ...result, stdout, stderr };
 }

src/mgr/store.ts

@@ -32,6 +32,20 @@ export interface DurableQueueClaim {
   attemptCount: number;
 }
 
+export interface RunnerRetentionFenceFacts {
+  run: RunRecord;
+  command: CommandRecord;
+  commands: CommandRecord[];
+  events: RunEvent[];
+}
+
+export interface RunnerRetentionFenceInput {
+  runId: string;
+  commandId: string;
+  terminalizeStalePending: boolean;
+  reason: string;
+}
+
 export interface AgentRunStore {
   health(): MaybePromise<StoreHealth>;
   createRun(input: CreateRunInput): MaybePromise<RunRecord>;
@@ -57,6 +71,8 @@ export interface AgentRunStore {
   getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): MaybePromise<RunnerJobRecord | null>;
   saveRunnerJob(input: SaveRunnerJobInput): MaybePromise<RunnerJobRecord>;
   updateRunnerJobResult(runnerJobId: string, patch: JsonRecord): MaybePromise<RunnerJobRecord>;
+  withRunnerCreateFence<T>(namespace: string, operation: () => Promise<T>): MaybePromise<T>;
+  withRunnerRetentionFence<T>(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise<T>): MaybePromise<T>;
   claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
   heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
   ackCommand(commandId: string): MaybePromise<CommandRecord>;
@@ -179,6 +195,8 @@ export class MemoryAgentRunStore implements AgentRunStore {
   private readonly kafkaEventOutbox = new Map<string, KafkaEventOutboxRecord>();
   private readonly queueTasks = new Map<string, QueueTaskRecord>();
   private readonly queueReadCursors = new Map<string, QueueReadCursorRecord>();
+  private readonly runnerCreateFenceTails = new Map<string, Promise<void>>();
+  private readonly runnerRetentionFences = new Set<string>();
   private queueVersion = 0;
   private sessionVersion = 0;
   private kafkaOutboxSeq = 0;
@@ -417,7 +435,40 @@ export class MemoryAgentRunStore implements AgentRunStore {
     return next;
   }
 
+  async withRunnerCreateFence<T>(namespace: string, operation: () => Promise<T>): Promise<T> {
+    const predecessor = this.runnerCreateFenceTails.get(namespace) ?? Promise.resolve();
+    let release = (): void => {};
+    const current = new Promise<void>((resolve) => { release = resolve; });
+    const tail = predecessor.then(async () => current);
+    this.runnerCreateFenceTails.set(namespace, tail);
+    await predecessor;
+    try {
+      return await operation();
+    } finally {
+      release();
+      if (this.runnerCreateFenceTails.get(namespace) === tail) this.runnerCreateFenceTails.delete(namespace);
+    }
+  }
+
+  async withRunnerRetentionFence<T>(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise<T>): Promise<T> {
+    if (this.runnerRetentionFences.has(input.runId)) throw runnerRetentionFenceConflict(input.runId);
+    const run = this.getRun(input.runId);
+    const command = this.getCommand(input.commandId);
+    if (command.runId !== run.id) throw new AgentRunError("schema-invalid", `command ${input.commandId} does not belong to run ${input.runId}`, { httpStatus: 409 });
+    this.runnerRetentionFences.add(input.runId);
+    try {
+      const commands = this.listCommands(run.id, 0, 100);
+      const events = this.listEvents(run.id, 0, 500);
+      const result = await operation({ run, command, commands, events });
+      if (input.terminalizeStalePending && !isTerminalRunStatus(run.status)) this.terminalizeRunnerRetention(run, commands, input.reason);
+      return result;
+    } finally {
+      this.runnerRetentionFences.delete(input.runId);
+    }
+  }
+
   claimRun(runId: string, runnerId: string, leaseMs: number): RunRecord {
+    if (this.runnerRetentionFences.has(runId)) throw runnerRetentionFenceConflict(runId);
     const run = this.getRun(runId);
     if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 });
     if (run.claimedBy && run.claimedBy !== runnerId && !isLeaseExpired(run.leaseExpiresAt)) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
@@ -428,6 +479,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
   }
 
   heartbeat(runId: string, runnerId: string, leaseMs: number): RunRecord {
+    if (this.runnerRetentionFences.has(runId)) throw runnerRetentionFenceConflict(runId);
     const run = this.getRun(runId);
     if (isTerminalRunStatus(run.status)) return run;
     if (run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is not claimed by ${runnerId}`, { httpStatus: 409 });
@@ -800,6 +852,20 @@ export class MemoryAgentRunStore implements AgentRunStore {
     return next;
   }
 
+  private terminalizeRunnerRetention(run: RunRecord, commands: CommandRecord[], reason: string): void {
+    const at = nowIso();
+    for (const command of commands.filter((item) => !isTerminalCommandState(item.state))) {
+      this.commands.set(command.id, { ...command, state: "failed", updatedAt: at });
+      this.cancelRunnerDispatchIntentForCommand(command.id, reason, at);
+      const prepared = this.prepareEvent(run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message: reason, retentionFence: true });
+      this.commitPreparedEvent(prepared);
+    }
+    const next = this.updateRun(run.id, { status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: reason, claimedBy: null, leaseExpiresAt: null });
+    this.cancelRunnerDispatchIntentsForRun(run.id, reason, at);
+    this.commitPreparedEvent(this.prepareEvent(run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message: reason, retentionFence: true }));
+    this.touchSessionForRun(next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at });
+  }
+
   private requireRunnerDispatchIntent(intentId: string): RunnerDispatchIntentRecord {
     const intent = this.runnerDispatchIntents.get(intentId);
     if (!intent) throw new AgentRunError("schema-invalid", `runner dispatch intent ${intentId} was not found`, { httpStatus: 404 });
@@ -1239,6 +1305,13 @@ function staleClaimError(component: string, claim: DurableQueueClaim): AgentRunE
   return new AgentRunError("runner-lease-conflict", `${component} claim is stale`, { httpStatus: 409, details: { id: claim.id, attemptCount: claim.attemptCount, valuesPrinted: false } });
 }
 
+function runnerRetentionFenceConflict(runId: string): AgentRunError {
+  return new AgentRunError("runner-lease-conflict", `run ${runId} is fenced by runner retention`, {
+    httpStatus: 409,
+    details: { reason: "runner-retention-db-fenced", runId, valuesPrinted: false },
+  });
+}
+
 function isActiveRunnerDispatchIntent(intent: RunnerDispatchIntentRecord): boolean {
   return intent.state === "pending" || intent.state === "dispatching" || intent.state === "retry";
 }

src/selftest/cases/20-runner-k8s-job.ts

@@ -299,6 +299,8 @@ function runnerJob(name, createdAt, identity) {
     metadata: {
       name,
       namespace: "agentrun-v02",
+      uid: "uid-" + name,
+      resourceVersion: "1",
       creationTimestamp: createdAt,
       labels: { "app.kubernetes.io/part-of": "agentrun", "app.kubernetes.io/name": "agentrun-runner", "app.kubernetes.io/component": "runner", "agentrun.pikastech.local/lane": "v0.2", "job-name": name },
       annotations: { "agentrun.pikastech.local/run-id": identity.runId, "agentrun.pikastech.local/command-id": identity.commandId, "agentrun.pikastech.local/runner-id": "runner-old", token: null, secret: null },
@@ -306,17 +308,27 @@ function runnerJob(name, createdAt, identity) {
     status: {},
   };
 }
+function retentionJobs() {
+  const facts = JSON.parse(process.env.AGENTRUN_SELFTEST_RETENTION_DB_FACTS || "[]");
+  return [
+    runnerJob("agentrun-v02-runner-old-a", "2026-01-01T00:00:00Z", facts[0]),
+    runnerJob("agentrun-v02-runner-old-b", "2026-01-01T00:01:00Z", facts[1]),
+  ];
+}
 if (args[0] === "get" && args[1] === "jobs") {
   const deleted = await readDeleted();
   const deletedJobs = new Set(deleted.filter((item) => item.resource === "job").map((item) => item.name));
-  const facts = JSON.parse(process.env.AGENTRUN_SELFTEST_RETENTION_DB_FACTS || "[]");
-  const items = [
-    runnerJob("agentrun-v02-runner-old-a", "2026-01-01T00:00:00Z", facts[0]),
-    runnerJob("agentrun-v02-runner-old-b", "2026-01-01T00:01:00Z", facts[1]),
-  ].filter((item) => !deletedJobs.has(item.metadata.name));
+  const items = retentionJobs().filter((item) => !deletedJobs.has(item.metadata.name));
   console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items }));
   process.exit(0);
 }
+if (args[0] === "get" && args[1] === "job" && args[2]) {
+  const deleted = await readDeleted();
+  const item = retentionJobs().find((candidate) => candidate.metadata.name === args[2]);
+  if (!item || deleted.some((entry) => entry.resource === "job" && entry.name === args[2])) process.exit(1);
+  console.log(JSON.stringify(item));
+  process.exit(0);
+}
 if (args[0] === "get" && args[1] === "pods") {
   console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] }));
   process.exit(0);
@@ -324,8 +336,15 @@ if (args[0] === "get" && args[1] === "pods") {
 if (args[0] === "get" && args[1] === "events") { console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: [] })); process.exit(0); }
 if (args[0] === "delete") {
   const deleted = await readDeleted();
-  const resource = args[1];
-  const name = args[2] && !args[2].startsWith("-") ? args[2] : args.join(" ");
+  const rawIndex = args.indexOf("--raw");
+  const rawPath = rawIndex >= 0 ? args[rawIndex + 1] : null;
+  const resource = rawPath?.includes("/jobs/") ? "job" : args[1];
+  const name = rawPath ? decodeURIComponent(rawPath.split("/").filter(Boolean).at(-1)) : args[2] && !args[2].startsWith("-") ? args[2] : args.join(" ");
+  if (rawPath) {
+    const body = JSON.parse(await readStdin());
+    const current = retentionJobs().find((candidate) => candidate.metadata.name === name);
+    if (!current || body?.preconditions?.uid !== current.metadata.uid || body?.preconditions?.resourceVersion !== current.metadata.resourceVersion) process.exit(1);
+  }
   deleted.push({ resource, name, args });
   await writeDeleted(deleted);
   console.log(String(resource) + "/" + String(name).replace(/[^a-z0-9-]+/g, "-") + " deleted");

src/selftest/cases/22-runner-retention-stale-pending.ts

@@ -4,7 +4,7 @@ import path from "node:path";
 import { AgentRunError } from "../../common/errors.js";
 import type { CommandRecord, JsonRecord, RunEvent, RunRecord } from "../../common/types.js";
 import { enforceRunnerRetentionBeforeCreate, type RunnerRetentionOptions, type RunnerRetentionSummary } from "../../mgr/runner-retention.js";
-import { MemoryAgentRunStore } from "../../mgr/store.js";
+import { MemoryAgentRunStore, type RunnerRetentionFenceFacts, type RunnerRetentionFenceInput } from "../../mgr/store.js";
 import type { SelfTestCase } from "../harness.js";
 
 const stalePendingMaxAgeMs = 15 * 60_000;
@@ -26,16 +26,21 @@ interface RetentionFixture {
   events?: JsonRecord[];
   eventsAfterFirstRead?: JsonRecord[];
   racePods?: JsonRecord[];
+  raceJobs?: JsonRecord[];
   eventsError?: boolean;
   trackEventReads?: boolean;
   churnEventMetadata?: boolean;
   raceAfterEvents?: boolean;
+  raceOnCasDelete?: boolean;
+  bumpJobResourceVersionAfterPodDelete?: boolean;
+  bumpedJobResourceVersion?: string;
   deleteErrorResources?: string[];
   retainDeletedJobs?: boolean;
 }
 
 class RetentionFactStore extends MemoryAgentRunStore {
   private readonly facts: RetentionFact[];
+  private beforeFence: (() => void) | null = null;
 
   constructor(facts: RetentionFact[]) {
     super();
@@ -64,6 +69,33 @@ class RetentionFactStore extends MemoryAgentRunStore {
     if (!fact) throw new AgentRunError("schema-invalid", `run ${runId} was not found`, { httpStatus: 404 });
     return fact.events.filter((event) => event.seq > afterSeq).slice(0, limit);
   }
+
+  setBeforeFence(callback: (() => void) | null): void {
+    this.beforeFence = callback;
+  }
+
+  claimFact(runId: string, runnerId: string): void {
+    const fact = this.facts.find((item) => item.run.id === runId);
+    if (!fact) throw new Error(`missing retention fact ${runId}`);
+    const at = new Date(Date.now() + 60_000).toISOString();
+    fact.run = { ...fact.run, status: "claimed", claimedBy: runnerId, leaseExpiresAt: at, updatedAt: at };
+    fact.events.push(runEvent(runId, fact.events.length + 1, at, "backend_status", { phase: "run-claimed", runnerId, commandId: fact.command.id }));
+  }
+
+  override async withRunnerRetentionFence<T>(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise<T>): Promise<T> {
+    const beforeFence = this.beforeFence;
+    this.beforeFence = null;
+    beforeFence?.();
+    const fact = this.facts.find((item) => item.run.id === input.runId && item.command.id === input.commandId);
+    if (!fact) throw new Error(`missing retention fence facts ${input.runId}/${input.commandId}`);
+    const result = await operation({ run: fact.run, command: fact.command, commands: this.listCommands(fact.run.id, 0, 100), events: this.listEvents(fact.run.id, 0, 500) });
+    if (input.terminalizeStalePending) {
+      const at = new Date().toISOString();
+      fact.command = { ...fact.command, state: "failed", updatedAt: at };
+      fact.run = { ...fact.run, status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: input.reason, claimedBy: null, leaseExpiresAt: null, updatedAt: at };
+    }
+    return result;
+  }
 }
 
 const selfTest: SelfTestCase = async (context) => {
@@ -81,8 +113,12 @@ const selfTest: SelfTestCase = async (context) => {
     await assertEventOnlyFailedMountCandidate({ fixturePath, statePath, kubectlCommand });
     await assertProtectedMatrix({ fixturePath, statePath, kubectlCommand });
     await assertPendingToRunningRaceFailsClosed({ fixturePath, statePath, kubectlCommand });
+    await assertDbClaimAfterRecheckFailsClosed({ fixturePath, statePath, kubectlCommand });
+    await assertPodCasRaceFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertDeleteFailureFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand });
+    await assertMemoryDbFenceRejectsConcurrentClaim();
+    await assertRunnerCreateFenceSerializesNamespace();
   } finally {
     restoreEnv("AGENTRUN_SELFTEST_RETENTION_FIXTURE_PATH", previousFixturePath);
     restoreEnv("AGENTRUN_SELFTEST_RETENTION_STATE_PATH", previousStatePath);
@@ -107,8 +143,12 @@ const selfTest: SelfTestCase = async (context) => {
       "event-facts-unavailable-protected",
       "terminal-ledger-running-pod-protected",
       "selected-runner-race-fails-closed",
-      "runner-delete-exit-code-checked",
+      "db-claim-after-recheck-fails-closed",
+      "pod-resource-version-cas-fails-closed",
+      "runner-delete-cas-exit-code-checked",
       "runner-delete-postcondition-checked",
+      "memory-db-fence-rejects-concurrent-claim",
+      "namespace-runner-create-fence-prevents-capacity-overshoot",
     ],
   };
 };
@@ -124,6 +164,8 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s
     events: facts.map((fact, index) => failedMountEvent(fact, index)),
     trackEventReads: true,
     churnEventMetadata: true,
+    bumpJobResourceVersionAfterPodDelete: true,
+    bumpedJobResourceVersion: "2",
   };
   await writeFixture(input, fixture);
   const store = new RetentionFactStore(facts);
@@ -163,6 +205,8 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s
   assert.equal(serialized.includes("REDACTED"), true);
   const deleted = await readDeletionState(input.statePath);
   assert.deepEqual(deleted.filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-00"]);
+  const jobCas = deleted.find((item) => item.resource === "cas-request" && item.targetKind === "job") ?? {};
+  assert.equal(jobCas.expectedResourceVersion, "2");
   assert.equal(deleted.filter((item) => item.resource === "read" && item.name === "events").length, 4);
 }
 
@@ -237,6 +281,43 @@ async function assertPendingToRunningRaceFailsClosed(input: { fixturePath: strin
   assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
 }
 
+async function assertDbClaimAfterRecheckFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(94, new Date(Date.now() - 60 * 60_000).toISOString());
+  await writeFixture(input, { jobs: [runnerJob(fact, 94)], pods: [runnerPod(fact, 94, "pending")], events: [failedMountEvent(fact, 94)] });
+  const store = new RetentionFactStore([fact]);
+  store.setBeforeFence(() => store.claimFact(fact.run.id, "runner-raced-after-recheck"));
+
+  const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ store, options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 }));
+  assert.equal(error.details?.reason, "runner-retention-db-fence-rejected");
+  assert.equal(error.details?.changeReason, "run-became-active");
+  assert.equal(error.details?.claimedBy, "runner-raced-after-recheck");
+  const state = await readDeletionState(input.statePath);
+  assert.equal(deletionOperations(state).length, 0);
+  assert.equal(state.filter((item) => item.resource === "cas-request").length, 0);
+}
+
+async function assertPodCasRaceFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(95, new Date(Date.now() - 60 * 60_000).toISOString());
+  const runningPod = runnerPod(fact, 95, "running");
+  (runningPod.metadata as JsonRecord).resourceVersion = "2";
+  await writeFixture(input, {
+    jobs: [runnerJob(fact, 95)],
+    pods: [runnerPod(fact, 95, "pending")],
+    events: [failedMountEvent(fact, 95)],
+    raceOnCasDelete: true,
+    racePods: [runningPod],
+  });
+
+  const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([fact]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 }));
+  assert.equal(error.details?.reason, "runner-retention-k8s-cas-rejected");
+  assert.equal(error.details?.changeReason, "object-identity-or-version-changed");
+  assert.equal(error.details?.resourceKind, "pod");
+  assert.equal(error.details?.expectedResourceVersion, "1");
+  const state = await readDeletionState(input.statePath);
+  assert.equal(deletionOperations(state).length, 0);
+  assert.equal(state.filter((item) => item.resource === "cas-request" && item.targetKind === "pod").length, 1);
+}
+
 async function assertDeleteFailureFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
   const fact = pendingFact(92, new Date(Date.now() - 60 * 60_000).toISOString());
   await writeFixture(input, {
@@ -251,10 +332,13 @@ async function assertDeleteFailureFailsClosed(input: { fixturePath: string; stat
     options: retentionOptions(input.kubectlCommand, 1),
     incomingRunnerCount: 1,
   }));
-  assert.equal(error.details?.reason, "runner-retention-delete-failed");
+  assert.equal(error.details?.reason, "runner-retention-k8s-cas-rejected");
+  assert.equal(error.details?.changeReason, "delete-request-failed");
   assert.equal(error.details?.resourceKind, "job");
   assert.equal(error.details?.exitCode, 42);
-  assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
+  const state = await readDeletionState(input.statePath);
+  assert.deepEqual(deletionOperations(state).filter((item) => item.resource === "job"), []);
+  assert.deepEqual(deletionOperations(state).filter((item) => item.resource === "pod").map((item) => item.name), ["agentrun-v02-runner-stale-92-pod-92"]);
 }
 
 async function assertDeletionPostconditionFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
@@ -279,6 +363,63 @@ async function assertDeletionPostconditionFailsClosed(input: { fixturePath: stri
   assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-93"]);
 }
 
+async function assertMemoryDbFenceRejectsConcurrentClaim(): Promise<void> {
+  const store = new MemoryAgentRunStore();
+  const run = store.createRun({ tenantId: "unidesk", projectId: "pikasTech/agentrun", workspaceRef: { kind: "host-path", path: "/tmp/agentrun-retention-memory" }, providerId: "NC01", backendProfile: "codex", executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } }, traceSink: null });
+  const command = store.createCommand(run.id, { type: "turn", payload: { prompt: "memory retention fence" } });
+  let enterFence = (): void => {};
+  let releaseFence = (): void => {};
+  const entered = new Promise<void>((resolve) => { enterFence = resolve; });
+  const release = new Promise<void>((resolve) => { releaseFence = resolve; });
+  const fenced = Promise.resolve(store.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "retention memory self-test" }, async () => {
+    enterFence();
+    await release;
+  }));
+  await entered;
+  assert.throws(
+    () => store.claimRun(run.id, "runner-during-retention", 60_000),
+    (error) => error instanceof AgentRunError && error.failureKind === "runner-lease-conflict" && error.details?.reason === "runner-retention-db-fenced",
+  );
+  releaseFence();
+  await fenced;
+  assert.equal(store.getRun(run.id).status, "failed");
+  assert.equal(store.getCommand(command.id).state, "failed");
+  assert.throws(() => store.claimRun(run.id, "runner-after-retention", 60_000), (error) => error instanceof AgentRunError && error.httpStatus === 409);
+}
+
+async function assertRunnerCreateFenceSerializesNamespace(): Promise<void> {
+  const store = new MemoryAgentRunStore();
+  const maxRunners = 20;
+  let liveRunnerCount = 19;
+  let enterFirst = (): void => {};
+  let releaseFirst = (): void => {};
+  const firstEntered = new Promise<void>((resolve) => { enterFirst = resolve; });
+  const firstRelease = new Promise<void>((resolve) => { releaseFirst = resolve; });
+  let secondEntered = false;
+  const first = Promise.resolve(store.withRunnerCreateFence("agentrun-nc01-v02", async () => {
+    assert.ok(liveRunnerCount + 1 <= maxRunners);
+    enterFirst();
+    await firstRelease;
+    liveRunnerCount++;
+    return "first";
+  }));
+  await firstEntered;
+  const second = Promise.resolve(store.withRunnerCreateFence("agentrun-nc01-v02", async () => {
+    secondEntered = true;
+    if (liveRunnerCount + 1 > maxRunners) throw new AgentRunError("infra-failed", "capacity fence rejected second incoming runner", { httpStatus: 409 });
+    liveRunnerCount++;
+    return "second";
+  }));
+  await Promise.resolve();
+  assert.equal(secondEntered, false);
+  releaseFirst();
+  const results = await Promise.allSettled([first, second]);
+  assert.equal(results[0]?.status, "fulfilled");
+  assert.equal(results[1]?.status, "rejected");
+  assert.equal(secondEntered, true);
+  assert.equal(liveRunnerCount, maxRunners);
+}
+
 async function assertProtected(
   input: { fixturePath: string; statePath: string; kubectlCommand: string },
   facts: RetentionFact[],

src/selftest/fixtures/fake-kubectl-retention-stale-pending.mjs

@@ -29,6 +29,25 @@ const eventReadCount = state.filter((item) => item.resource === "read" && item.n
 const deletedJobs = fixture.retainDeletedJobs === true
   ? new Set()
   : new Set(state.filter((item) => item.resource === "job").map((item) => item.name));
+const deletedPods = new Set(state.filter((item) => item.resource === "pod").map((item) => item.name));
+
+function currentJob(name) {
+  const base = fixture.jobs.find((item) => item?.metadata?.name === name);
+  if (!base || fixture.bumpJobResourceVersionAfterPodDelete !== true) return base;
+  const podDeleted = fixture.pods.some((pod) => podJobName(pod) === name && deletedPods.has(pod?.metadata?.name));
+  if (!podDeleted) return base;
+  return { ...base, metadata: { ...(base.metadata ?? {}), resourceVersion: fixture.bumpedJobResourceVersion ?? "2" } };
+}
+
+if (args[0] === "get" && args[1] === "job" && args[2]) {
+  const item = currentJob(String(args[2]));
+  if (!item || deletedJobs.has(item?.metadata?.name)) {
+    console.error(`Error from server (NotFound): job/${String(args[2])} was not found`);
+    process.exit(1);
+  }
+  console.log(JSON.stringify(item));
+  process.exit(0);
+}
 
 if (args[0] === "get" && args[1] === "jobs") {
   const items = fixture.jobs.filter((item) => !deletedJobs.has(item?.metadata?.name));
@@ -40,7 +59,7 @@ if (args[0] === "get" && args[1] === "pods") {
   const sourcePods = fixture.raceAfterEvents === true && eventReadCount > 0 && Array.isArray(fixture.racePods)
     ? fixture.racePods
     : fixture.pods;
-  const items = sourcePods.filter((item) => !deletedJobs.has(podJobName(item)));
+  const items = sourcePods.filter((item) => !deletedJobs.has(podJobName(item)) && !deletedPods.has(item?.metadata?.name));
   console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items }));
   process.exit(0);
 }
@@ -70,6 +89,32 @@ if (args[0] === "get" && args[1] === "events") {
 }
 
 if (args[0] === "delete") {
+  const rawIndex = args.indexOf("--raw");
+  if (rawIndex >= 0) {
+    const resourcePath = String(args[rawIndex + 1] ?? "");
+    const resource = resourcePath.includes("/jobs/") ? "job" : resourcePath.includes("/pods/") ? "pod" : "unknown";
+    const name = decodeURIComponent(resourcePath.split("/").filter(Boolean).at(-1) ?? "unknown");
+    const deleteOptions = await readStdinJson();
+    const expected = deleteOptions?.preconditions ?? {};
+    const source = fixture.raceOnCasDelete === true && Array.isArray(fixture.racePods) ? fixture.racePods : fixture.pods;
+    const current = resource === "job"
+      ? (fixture.raceOnCasDelete === true && Array.isArray(fixture.raceJobs) ? fixture.raceJobs.find((item) => item?.metadata?.name === name) : currentJob(name))
+      : source.find((item) => item?.metadata?.name === name);
+    state.push({ resource: "cas-request", name, targetKind: resource, expectedUid: expected.uid ?? null, expectedResourceVersion: expected.resourceVersion ?? null, valuesPrinted: false });
+    await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`);
+    if (!expected.uid || !expected.resourceVersion || !current || current?.metadata?.uid !== expected.uid || current?.metadata?.resourceVersion !== expected.resourceVersion) {
+      console.error(`Error from server (Conflict): ${resource}/${name} resourceVersion or UID precondition failed`);
+      process.exit(1);
+    }
+    if (Array.isArray(fixture.deleteErrorResources) && fixture.deleteErrorResources.includes(resource)) {
+      console.error(`self-test forced ${resource} deletion failure`);
+      process.exit(42);
+    }
+    state.push({ resource, name, selector: null, cas: true, valuesPrinted: false });
+    await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`);
+    console.log(JSON.stringify({ kind: "Status", apiVersion: "v1", status: "Success", details: { name, kind: resource } }));
+    process.exit(0);
+  }
   const resource = String(args[1] ?? "unknown");
   const directName = args[2] && !String(args[2]).startsWith("-") ? String(args[2]) : null;
   const selectorIndex = args.indexOf("-l");
@@ -85,5 +130,13 @@ if (args[0] === "delete") {
   process.exit(0);
 }
 
+async function readStdinJson() {
+  try {
+    return JSON.parse(await Bun.stdin.text());
+  } catch {
+    return null;
+  }
+}
+
 console.error(`unsupported fake retention kubectl args: ${args.join(" ")}`);
 process.exit(1);

src/selftest/integration/postgres-runner-retention-fence.ts

@@ -0,0 +1,80 @@
+import assert from "node:assert/strict";
+import { AgentRunError } from "../../common/errors.js";
+import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js";
+
+const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim();
+if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory");
+
+const primary = await createPostgresAgentRunStore({ connectionString });
+const concurrent = await createPostgresAgentRunStore({ connectionString });
+
+try {
+  await assertNamespaceCreateFence();
+  await assertRetentionFenceBlocksClaim();
+  console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-run-retention-fence-blocks-claim"] }));
+} finally {
+  await Promise.all([primary.close(), concurrent.close()]);
+}
+
+async function assertNamespaceCreateFence(): Promise<void> {
+  let firstEntered = (): void => {};
+  let releaseFirst = (): void => {};
+  const entered = new Promise<void>((resolve) => { firstEntered = resolve; });
+  const release = new Promise<void>((resolve) => { releaseFirst = resolve; });
+  let secondEntered = false;
+  const first = primary.withRunnerCreateFence("agentrun-postgres-retention-selftest", async () => {
+    firstEntered();
+    await release;
+    return "first";
+  });
+  await entered;
+  const second = concurrent.withRunnerCreateFence("agentrun-postgres-retention-selftest", async () => {
+    secondEntered = true;
+    return "second";
+  });
+  await delay(50);
+  assert.equal(secondEntered, false);
+  releaseFirst();
+  assert.deepEqual(await Promise.all([first, second]), ["first", "second"]);
+}
+
+async function assertRetentionFenceBlocksClaim(): Promise<void> {
+  const run = await primary.createRun({
+    tenantId: "selftest",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-retention-selftest" },
+    providerId: "NC01",
+    backendProfile: "codex",
+    executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } },
+    traceSink: null,
+  });
+  const command = await primary.createCommand(run.id, { type: "turn", payload: { prompt: "postgres retention fence self-test" } });
+  let fenceEntered = (): void => {};
+  let releaseFence = (): void => {};
+  const entered = new Promise<void>((resolve) => { fenceEntered = resolve; });
+  const release = new Promise<void>((resolve) => { releaseFence = resolve; });
+  const fenced = primary.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "postgres retention fence self-test" }, async (facts) => {
+    assert.equal(facts.run.id, run.id);
+    assert.equal(facts.command.id, command.id);
+    fenceEntered();
+    await release;
+  });
+  await entered;
+  let claimSettled = false;
+  const claim = concurrent.claimRun(run.id, "runner-postgres-retention-race", 60_000)
+    .then(() => null, (error: unknown) => error)
+    .finally(() => { claimSettled = true; });
+  await delay(50);
+  assert.equal(claimSettled, false);
+  releaseFence();
+  await fenced;
+  const claimError = await claim;
+  assert.ok(claimError instanceof AgentRunError);
+  assert.equal(claimError.httpStatus, 409);
+  assert.equal((await primary.getRun(run.id)).status, "failed");
+  assert.equal((await primary.getCommand(command.id)).state, "failed");
+}
+
+async function delay(ms: number): Promise<void> {
+  await new Promise<void>((resolve) => setTimeout(resolve, ms));
+}