fix: 关闭 runner retention 删除竞态

src/mgr/runner-retention.ts

@@ -50,6 +50,8 @@ type CandidateKind = "idle" | "terminal" | "inactive" | "stale-pending";
 
 interface RunnerPodObservation extends JsonRecord {
   name: string;
+  uid: string | null;
+  resourceVersion: string | null;
   phase: string | null;
   createdAt: string | null;
   ageMs: number | null;
@@ -57,6 +59,7 @@ interface RunnerPodObservation extends JsonRecord {
   podStartTime: string | null;
   ready: boolean;
   runnerContainerState: "waiting" | "running" | "terminated" | "unknown";
+  runnerContainerLastState: "running" | "terminated" | "none";
   runnerContainerStarted: boolean;
   runnerContainerReady: boolean;
   runnerContainerRestartCount: number;
@@ -72,6 +75,8 @@ interface K8sObject {
   metadata?: {
     name?: string;
     namespace?: string;
+    uid?: string;
+    resourceVersion?: string;
     creationTimestamp?: string;
     labels?: Record<string, string>;
     annotations?: Record<string, string>;
@@ -93,6 +98,8 @@ interface RunnerResourceEntry {
   resourceKind: "job" | "pod";
   name: string;
   jobName: string;
+  resourceUid: string | null;
+  resourceVersion: string | null;
   runId: string | null;
   commandId: string | null;
   runnerId: string | null;
@@ -105,6 +112,7 @@ interface RunnerResourceEntry {
   candidateKind: CandidateKind;
   classificationReason: string;
   lastActiveAtMs: number;
+  dbFactsFingerprint: string | null;
 }
 
 interface Snapshot {
@@ -157,16 +165,37 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
   let deletedRunnerPodCount = 0;
   let deletedAssociatedResourceCount = 0;
   for (const item of selected) {
+    await revalidateSelectedRunner(input.store, input.options, item);
     if (item.resourceKind === "job") {
-      await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "job", item.name, "-n", input.options.namespace, "--ignore-not-found=true"], input.options.signal);
+      await deleteRunnerResource(input.options, item);
       deletedRunnerJobCount++;
       deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
     } else {
-      await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "pod", item.name, "-n", input.options.namespace, "--ignore-not-found=true"], input.options.signal);
+      await deleteRunnerResource(input.options, item);
       deletedRunnerPodCount++;
     }
   }
   const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before;
+  if (after.liveRunnerCount > targetBeforeCreate) {
+    throw new AgentRunError("infra-failed", "runner retention did not make enough room for a new runner", {
+      httpStatus: 409,
+      details: {
+        reason: "runner-retention-postcondition-failed",
+        namespace: input.options.namespace,
+        maxRunners: input.options.maxRunners,
+        incomingRunnerCount,
+        targetBeforeCreate,
+        liveRunnerCountBefore: before.liveRunnerCount,
+        liveRunnerCountAfter: after.liveRunnerCount,
+        selectedRunnerCount: selected.length,
+        deletedRunnerJobCount,
+        deletedRunnerPodCount,
+        deletedAssociatedResourceCount,
+        selected: selected.map(resourceSummary),
+        valuesPrinted: false,
+      },
+    });
+  }
   return {
     enabled: true,
     namespace: input.options.namespace,
@@ -287,6 +316,8 @@ async function jobEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe
     resourceKind: "job",
     name,
     jobName: name,
+    resourceUid: item.metadata?.uid ?? null,
+    resourceVersion: item.metadata?.resourceVersion ?? null,
     createdAtMs: objectCreatedAtMs(item),
     terminal: jobTerminal(item),
     podPhase: null,
@@ -314,6 +345,8 @@ async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe
     resourceKind: "pod",
     name,
     jobName,
+    resourceUid: item.metadata?.uid ?? null,
+    resourceVersion: item.metadata?.resourceVersion ?? null,
     createdAtMs: objectCreatedAtMs(item),
     terminal: podTerminal(item),
     podPhase: stringPath(item, ["status", "phase"]),
@@ -322,7 +355,7 @@ async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe
   };
 }
 
-type ActivityFields = Pick<RunnerResourceEntry, "runId" | "commandId" | "runnerId" | "protectedActive" | "protectedReason" | "candidateKind" | "classificationReason" | "lastActiveAtMs">;
+type ActivityFields = Pick<RunnerResourceEntry, "runId" | "commandId" | "runnerId" | "protectedActive" | "protectedReason" | "candidateKind" | "classificationReason" | "lastActiveAtMs" | "dbFactsFingerprint">;
 
 async function activityFor(store: AgentRunStore, input: {
   runId: string | null;
@@ -335,41 +368,85 @@ async function activityFor(store: AgentRunStore, input: {
   podObservations: RunnerPodObservation[];
   podIdentityMatch: boolean;
 }): Promise<ActivityFields> {
-  if (!input.runId || !input.commandId || !input.runnerId) return protectedActivity(input, "identity-facts-unavailable", input.createdAtMs);
-  if (!input.podIdentityMatch) return protectedActivity(input, "pod-identity-mismatch", input.createdAtMs);
+  if (!input.runId || !input.commandId || !input.runnerId) return protectedActivity(input, "identity-facts-unavailable", input.createdAtMs, null);
+  if (!input.podIdentityMatch) return protectedActivity(input, "pod-identity-mismatch", input.createdAtMs, null);
 
   let run: RunRecord;
   let command: CommandRecord;
   try {
     [run, command] = await Promise.all([store.getRun(input.runId), store.getCommand(input.commandId)]);
   } catch {
-    return protectedActivity(input, "db-facts-unavailable", input.createdAtMs);
+    return protectedActivity(input, "db-facts-unavailable", input.createdAtMs, null);
   }
 
   const lastActiveAtMs = Math.max(input.createdAtMs, isoMs(run.updatedAt), isoMs(command.updatedAt), isoMs(run.leaseExpiresAt));
-  if (command.runId !== run.id) return protectedActivity(input, "db-identity-mismatch", lastActiveAtMs);
-  const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs));
-  if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs);
-  const activePodReason = activePodProtectionReason(input.podObservations);
-  if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs);
-  if (isTerminalRunStatus(run.status)) return candidateActivity(input, "terminal", "terminal-run", lastActiveAtMs);
-  if (isTerminalCommandState(command.state)) return candidateActivity(input, input.terminal ? "terminal" : "idle", input.terminal ? "terminal-runner-resource" : "terminal-command-idle-resource", lastActiveAtMs);
-  if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs);
-
-  const podBlockReason = stalePendingPodBlockReason(input.podObservations);
-  if (podBlockReason) return protectedActivity(input, podBlockReason, lastActiveAtMs);
-
+  if (command.runId !== run.id) return protectedActivity(input, "db-identity-mismatch", lastActiveAtMs, null);
   let commands: CommandRecord[];
   let events: import("../common/types.js").RunEvent[];
   try {
     [commands, events] = await Promise.all([store.listCommands(run.id, 0, 100), store.listEvents(run.id, 0, 500)]);
   } catch {
-    return protectedActivity(input, "db-progress-facts-unavailable", lastActiveAtMs);
+    return protectedActivity(input, "db-progress-facts-unavailable", lastActiveAtMs, null);
   }
+  const dbFactsFingerprint = databaseFactsFingerprint(run, command, commands, events);
+  const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs));
+  if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs, dbFactsFingerprint);
+  const activePodReason = activePodProtectionReason(input.podObservations);
+  if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs, dbFactsFingerprint);
+  if (isTerminalRunStatus(run.status)) return candidateActivity(input, "terminal", "terminal-run", lastActiveAtMs, dbFactsFingerprint);
+  if (isTerminalCommandState(command.state)) return candidateActivity(input, input.terminal ? "terminal" : "idle", input.terminal ? "terminal-runner-resource" : "terminal-command-idle-resource", lastActiveAtMs, dbFactsFingerprint);
+  if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs, dbFactsFingerprint);
+
+  const podBlockReason = stalePendingPodBlockReason(input.podObservations);
+  if (podBlockReason) return protectedActivity(input, podBlockReason, lastActiveAtMs, dbFactsFingerprint);
   if (!hasNoBusinessProgress(run, command, commands, events, input.stalePendingMaxAgeMs)) {
-    return protectedActivity(input, "business-progress-observed-or-unverifiable", lastActiveAtMs);
+    return protectedActivity(input, "business-progress-observed-or-unverifiable", lastActiveAtMs, dbFactsFingerprint);
   }
-  return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs);
+  return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint);
+}
+
+async function revalidateSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, selected: RunnerResourceEntry): Promise<void> {
+  const snapshot = await runnerSnapshot(store, options);
+  const resources = selected.resourceKind === "job" ? snapshot.jobs : snapshot.pods;
+  const current = resources.find((item) => item.name === selected.name) ?? null;
+  const selectedFingerprint = runnerSafetyFingerprint(selected);
+  const currentFingerprint = current ? runnerSafetyFingerprint(current) : null;
+  if (current && !current.protectedActive && currentFingerprint === selectedFingerprint) return;
+
+  throw new AgentRunError("infra-failed", "runner retention selected resource changed before deletion", {
+    httpStatus: 409,
+    details: {
+      reason: "runner-retention-selected-runner-changed",
+      changeReason: current === null ? "resource-missing" : current.protectedActive ? "protected-active" : "safety-fingerprint-changed",
+      namespace: options.namespace,
+      resourceKind: selected.resourceKind,
+      name: selected.name,
+      selectedFingerprint,
+      currentFingerprint,
+      selected: resourceSummary(selected),
+      current: current ? resourceSummary(current) : null,
+      valuesPrinted: false,
+    },
+  });
+}
+
+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",
+      namespace: options.namespace,
+      resourceKind: item.resourceKind,
+      name: item.name,
+      exitCode: result.code,
+      signal: result.signal,
+      stderr: redactText(result.stderr.slice(-2000)),
+      stdout: redactText(result.stdout.slice(-1000)),
+      valuesPrinted: false,
+    }),
+  });
 }
 
 async function kubectlGetList(options: RunnerRetentionOptions, resource: string, selector: string): Promise<K8sObject[]> {
@@ -430,6 +507,8 @@ function resourceSummary(item: RunnerResourceEntry): JsonRecord {
     resourceKind: item.resourceKind,
     name: item.name,
     jobName: item.jobName,
+    resourceUid: item.resourceUid,
+    resourceVersion: item.resourceVersion,
     runId: item.runId,
     commandId: item.commandId,
     runnerId: item.runnerId,
@@ -510,10 +589,13 @@ function observePod(item: K8sObject, stalePendingMaxAgeMs: number, events: K8sOb
   ];
   const runnerContainerRestartCount = typeof runner?.restartCount === "number" ? runner.restartCount : 0;
   const runnerLastState = jsonRecord(runner?.lastState);
+  const runnerContainerLastState: RunnerPodObservation["runnerContainerLastState"] = runnerLastState?.running ? "running" : runnerLastState?.terminated ? "terminated" : "none";
   const runnerContainerStarted = runner?.started === true || runnerContainerRestartCount > 0 || runnerContainerState === "running" || runnerContainerState === "terminated" || Boolean(runnerLastState?.running) || Boolean(runnerLastState?.terminated);
   const runnerContainerReady = runner?.ready === true;
   return {
     name: objectName(item) ?? "unknown",
+    uid: item.metadata?.uid ?? null,
+    resourceVersion: item.metadata?.resourceVersion ?? null,
     phase,
     createdAt: createdAtMs > 0 ? new Date(createdAtMs).toISOString() : null,
     ageMs,
@@ -521,6 +603,7 @@ function observePod(item: K8sObject, stalePendingMaxAgeMs: number, events: K8sOb
     podStartTime: boundedRedactedText(item.status?.startTime),
     ready,
     runnerContainerState,
+    runnerContainerLastState,
     runnerContainerStarted,
     runnerContainerReady,
     runnerContainerRestartCount,
@@ -591,12 +674,117 @@ function hasNoBusinessProgress(run: RunRecord, command: CommandRecord, commands:
   return events.every((event) => event.payload.commandId === undefined || event.payload.commandId === command.id);
 }
 
-function protectedActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, reason: string, lastActiveAtMs: number): ActivityFields {
-  return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: reason, candidateKind: "inactive", classificationReason: reason, lastActiveAtMs };
+function protectedActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string | null): ActivityFields {
+  return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: reason, candidateKind: "inactive", classificationReason: reason, lastActiveAtMs, dbFactsFingerprint };
 }
 
-function candidateActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, kind: CandidateKind, reason: string, lastActiveAtMs: number): ActivityFields {
-  return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind: kind, classificationReason: reason, lastActiveAtMs };
+function candidateActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, kind: CandidateKind, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string): ActivityFields {
+  return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind: kind, classificationReason: reason, lastActiveAtMs, dbFactsFingerprint };
+}
+
+function databaseFactsFingerprint(run: RunRecord, command: CommandRecord, commands: CommandRecord[], events: import("../common/types.js").RunEvent[]): string {
+  const commandFacts = (item: CommandRecord): JsonRecord => ({
+    id: item.id,
+    runId: item.runId,
+    seq: item.seq,
+    type: item.type,
+    state: item.state,
+    payloadHash: item.payloadHash,
+    cancelEpoch: item.cancelEpoch,
+    cancelRequestId: item.cancelRequestId,
+    cancelRequestedAt: item.cancelRequestedAt,
+    cancelReason: item.cancelReason,
+    createdAt: item.createdAt,
+    updatedAt: item.updatedAt,
+    acknowledgedAt: item.acknowledgedAt,
+    dispatchIntentHash: stableHash(item.dispatchIntent ?? null),
+    valuesPrinted: false,
+  });
+  return stableHash({
+    run: {
+      id: run.id,
+      status: run.status,
+      terminalStatus: run.terminalStatus,
+      failureKind: run.failureKind,
+      failureMessageHash: run.failureMessage ? stableHash(run.failureMessage) : null,
+      cancelEpoch: run.cancelEpoch,
+      cancelRequestId: run.cancelRequestId,
+      cancelRequestedAt: run.cancelRequestedAt,
+      cancelReason: run.cancelReason,
+      createdAt: run.createdAt,
+      updatedAt: run.updatedAt,
+      claimedBy: run.claimedBy,
+      leaseExpiresAt: run.leaseExpiresAt,
+      valuesPrinted: false,
+    },
+    command: commandFacts(command),
+    commands: [...commands].sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)).map(commandFacts),
+    events: [...events].sort((left, right) => left.seq - right.seq || left.id.localeCompare(right.id)).map((event) => ({
+      id: event.id,
+      runId: event.runId,
+      seq: event.seq,
+      type: event.type,
+      createdAt: event.createdAt,
+      payloadHash: stableHash(event.payload),
+      valuesPrinted: false,
+    })),
+    valuesPrinted: false,
+  });
+}
+
+function runnerSafetyFingerprint(item: RunnerResourceEntry): string {
+  return stableHash({
+    schema: "agentrun-runner-retention-safety-v1",
+    resourceKind: item.resourceKind,
+    name: item.name,
+    jobName: item.jobName,
+    resourceUid: item.resourceUid,
+    resourceVersion: item.resourceVersion,
+    runId: item.runId,
+    commandId: item.commandId,
+    runnerId: item.runnerId,
+    createdAtMs: item.createdAtMs,
+    terminal: item.terminal,
+    podPhase: item.podPhase,
+    protectedActive: item.protectedActive,
+    protectedReason: item.protectedReason,
+    candidateKind: item.candidateKind,
+    classificationReason: item.classificationReason,
+    lastActiveAtMs: item.lastActiveAtMs,
+    dbFactsFingerprint: item.dbFactsFingerprint,
+    podObservations: [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name)).map(podSafetyFacts),
+    valuesPrinted: false,
+  });
+}
+
+function podSafetyFacts(item: RunnerPodObservation): JsonRecord {
+  const waiting = item.waiting.map((entry) => ({
+    source: typeof entry.source === "string" ? entry.source : null,
+    name: typeof entry.name === "string" ? entry.name : null,
+    reason: typeof entry.reason === "string" ? entry.reason : null,
+    deterministic: entry.deterministic === true,
+    valuesPrinted: false,
+  })).sort((left, right) => JSON.stringify(left).localeCompare(JSON.stringify(right)));
+  return {
+    name: item.name,
+    uid: item.uid,
+    resourceVersion: item.resourceVersion,
+    phase: item.phase,
+    createdAt: item.createdAt,
+    stale: item.stale,
+    podStartTime: item.podStartTime,
+    ready: item.ready,
+    runnerContainerState: item.runnerContainerState,
+    runnerContainerLastState: item.runnerContainerLastState,
+    runnerContainerStarted: item.runnerContainerStarted,
+    runnerContainerReady: item.runnerContainerReady,
+    runnerContainerRestartCount: item.runnerContainerRestartCount,
+    waiting,
+    waitingEvidenceSources: [...item.waitingEvidenceSources].sort(),
+    eventFactsAvailable: item.eventFactsAvailable,
+    factsAvailable: item.factsAvailable,
+    valuesPrinted: false,
+  };
 }
 
 function boundedResourceSummaries(items: RunnerResourceEntry[]): { items: JsonRecord[]; omittedCount: number } {

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

@@ -20,6 +20,20 @@ interface RetentionFact {
   events: RunEvent[];
 }
 
+interface RetentionFixture {
+  jobs: JsonRecord[];
+  pods: JsonRecord[];
+  events?: JsonRecord[];
+  eventsAfterFirstRead?: JsonRecord[];
+  racePods?: JsonRecord[];
+  eventsError?: boolean;
+  trackEventReads?: boolean;
+  churnEventMetadata?: boolean;
+  raceAfterEvents?: boolean;
+  deleteErrorResources?: string[];
+  retainDeletedJobs?: boolean;
+}
+
 class RetentionFactStore extends MemoryAgentRunStore {
   private readonly facts: RetentionFact[];
 
@@ -66,6 +80,9 @@ const selfTest: SelfTestCase = async (context) => {
     await assertTwentyStalePendingSelectsOnlyOldest({ fixturePath, statePath, kubectlCommand });
     await assertEventOnlyFailedMountCandidate({ fixturePath, statePath, kubectlCommand });
     await assertProtectedMatrix({ fixturePath, statePath, kubectlCommand });
+    await assertPendingToRunningRaceFailsClosed({ fixturePath, statePath, kubectlCommand });
+    await assertDeleteFailureFailsClosed({ fixturePath, statePath, kubectlCommand });
+    await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand });
   } finally {
     restoreEnv("AGENTRUN_SELFTEST_RETENTION_FIXTURE_PATH", previousFixturePath);
     restoreEnv("AGENTRUN_SELFTEST_RETENTION_STATE_PATH", previousStatePath);
@@ -89,6 +106,9 @@ const selfTest: SelfTestCase = async (context) => {
       "pod-identity-mismatch-protected",
       "event-facts-unavailable-protected",
       "terminal-ledger-running-pod-protected",
+      "selected-runner-race-fails-closed",
+      "runner-delete-exit-code-checked",
+      "runner-delete-postcondition-checked",
     ],
   };
 };
@@ -102,6 +122,8 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s
     jobs: facts.map((fact, index) => runnerJob(fact, index)),
     pods: facts.map((fact, index) => runnerPod(fact, index, "pending")),
     events: facts.map((fact, index) => failedMountEvent(fact, index)),
+    trackEventReads: true,
+    churnEventMetadata: true,
   };
   await writeFixture(input, fixture);
   const store = new RetentionFactStore(facts);
@@ -111,7 +133,7 @@ async function assertTwentyStalePendingSelectsOnlyOldest(input: { fixturePath: s
   assert.equal(withinLimit.liveRunnerCountBefore, 20);
   assert.equal(withinLimit.inactiveCandidateCount, 20);
   assert.equal(withinLimit.selectedRunnerCount, 0);
-  assert.equal((await readDeletionState(input.statePath)).length, 0);
+  assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
 
   const cleanup = await enforceRunnerRetentionBeforeCreate({ store, options, incomingRunnerCount: 1 });
   assert.equal(cleanup.liveRunnerCountBefore, 20);
@@ -141,6 +163,7 @@ 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"]);
+  assert.equal(deleted.filter((item) => item.resource === "read" && item.name === "events").length, 4);
 }
 
 async function assertProtectedMatrix(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
@@ -190,6 +213,72 @@ async function assertEventOnlyFailedMountCandidate(input: { fixturePath: string;
   assert.equal((await readDeletionState(input.statePath)).length, 0);
 }
 
+async function assertPendingToRunningRaceFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(91, new Date(Date.now() - 60 * 60_000).toISOString());
+  const runningPod = runnerPod(fact, 91, "running");
+  (runningPod.metadata as JsonRecord).resourceVersion = "2";
+  await writeFixture(input, {
+    jobs: [runnerJob(fact, 91)],
+    pods: [runnerPod(fact, 91, "pending")],
+    events: [failedMountEvent(fact, 91)],
+    raceAfterEvents: 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-selected-runner-changed");
+  assert.equal(error.details?.changeReason, "protected-active");
+  const current = (error.details?.current ?? {}) as JsonRecord;
+  assert.equal(current.protectedReason, "runner-pod-started-running-or-ready");
+  assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
+}
+
+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, {
+    jobs: [runnerJob(fact, 92)],
+    pods: [runnerPod(fact, 92, "pending")],
+    events: [failedMountEvent(fact, 92)],
+    deleteErrorResources: ["job"],
+  });
+
+  const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({
+    store: new RetentionFactStore([fact]),
+    options: retentionOptions(input.kubectlCommand, 1),
+    incomingRunnerCount: 1,
+  }));
+  assert.equal(error.details?.reason, "runner-retention-delete-failed");
+  assert.equal(error.details?.resourceKind, "job");
+  assert.equal(error.details?.exitCode, 42);
+  assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
+}
+
+async function assertDeletionPostconditionFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(93, new Date(Date.now() - 60 * 60_000).toISOString());
+  await writeFixture(input, {
+    jobs: [runnerJob(fact, 93)],
+    pods: [runnerPod(fact, 93, "pending")],
+    events: [failedMountEvent(fact, 93)],
+    retainDeletedJobs: true,
+  });
+
+  const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({
+    store: new RetentionFactStore([fact]),
+    options: retentionOptions(input.kubectlCommand, 1),
+    incomingRunnerCount: 1,
+  }));
+  assert.equal(error.details?.reason, "runner-retention-postcondition-failed");
+  assert.equal(error.details?.targetBeforeCreate, 0);
+  assert.equal(error.details?.liveRunnerCountBefore, 1);
+  assert.equal(error.details?.liveRunnerCountAfter, 1);
+  assert.equal(error.details?.deletedRunnerJobCount, 1);
+  assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-93"]);
+}
+
 async function assertProtected(
   input: { fixturePath: string; statePath: string; kubectlCommand: string },
   facts: RetentionFact[],
@@ -287,6 +376,8 @@ function runnerJob(fact: RetentionFact, index: number): JsonRecord {
     metadata: {
       name,
       namespace: "agentrun-nc01-v02",
+      uid: `job-uid-${suffix}`,
+      resourceVersion: "1",
       creationTimestamp: fact.run.createdAt,
       labels: { ...labels, "job-name": name },
       annotations: identityAnnotations(fact),
@@ -319,6 +410,8 @@ function runnerPod(fact: RetentionFact, index: number, state: "pending" | "runni
     metadata: {
       name,
       namespace: "agentrun-nc01-v02",
+      uid: `pod-uid-${suffix}`,
+      resourceVersion: "1",
       creationTimestamp: fact.run.createdAt,
       labels: { ...labels, "job-name": jobName },
       annotations: identityAnnotations(fact),
@@ -372,7 +465,7 @@ function retentionOptions(kubectlCommand: string, maxRunners: number): RunnerRet
   };
 }
 
-async function writeFixture(input: { fixturePath: string; statePath: string }, fixture: { jobs: JsonRecord[]; pods: JsonRecord[]; events?: JsonRecord[]; eventsError?: boolean }): Promise<void> {
+async function writeFixture(input: { fixturePath: string; statePath: string }, fixture: RetentionFixture): Promise<void> {
   await writeFile(input.fixturePath, `${JSON.stringify(fixture, null, 2)}\n`, "utf8");
   await writeFile(input.statePath, "[]\n", "utf8");
 }
@@ -381,6 +474,10 @@ async function readDeletionState(statePath: string): Promise<JsonRecord[]> {
   return JSON.parse(await readFile(statePath, "utf8")) as JsonRecord[];
 }
 
+function deletionOperations(state: JsonRecord[]): JsonRecord[] {
+  return state.filter((item) => item.resource === "job" || item.resource === "pod" || item.resource === "secret" || item.resource === "pvc");
+}
+
 async function retentionError(run: () => Promise<RunnerRetentionSummary>): Promise<AgentRunError> {
   try {
     await run();

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

@@ -25,7 +25,10 @@ function podJobName(item) {
 
 const fixture = await readJson(fixturePath, { jobs: [], pods: [] });
 const state = await readJson(statePath, []);
-const deletedJobs = new Set(state.filter((item) => item.resource === "job").map((item) => item.name));
+const eventReadCount = state.filter((item) => item.resource === "read" && item.name === "events").length;
+const deletedJobs = fixture.retainDeletedJobs === true
+  ? new Set()
+  : new Set(state.filter((item) => item.resource === "job").map((item) => item.name));
 
 if (args[0] === "get" && args[1] === "jobs") {
   const items = fixture.jobs.filter((item) => !deletedJobs.has(item?.metadata?.name));
@@ -34,7 +37,10 @@ if (args[0] === "get" && args[1] === "jobs") {
 }
 
 if (args[0] === "get" && args[1] === "pods") {
-  const items = fixture.pods.filter((item) => !deletedJobs.has(podJobName(item)));
+  const sourcePods = fixture.raceAfterEvents === true && eventReadCount > 0 && Array.isArray(fixture.racePods)
+    ? fixture.racePods
+    : fixture.pods;
+  const items = sourcePods.filter((item) => !deletedJobs.has(podJobName(item)));
   console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items }));
   process.exit(0);
 }
@@ -44,7 +50,22 @@ if (args[0] === "get" && args[1] === "events") {
     console.error("events access denied by self-test fixture");
     process.exit(1);
   }
-  console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items: fixture.events ?? [] }));
+  const sourceItems = eventReadCount > 0 && Array.isArray(fixture.eventsAfterFirstRead)
+    ? fixture.eventsAfterFirstRead
+    : fixture.events ?? [];
+  const items = fixture.churnEventMetadata === true
+    ? sourceItems.map((item) => ({
+      ...item,
+      count: Number(item?.count ?? 0) + eventReadCount,
+      lastTimestamp: new Date(Date.parse(item?.lastTimestamp ?? "2026-01-01T00:00:00.000Z") + eventReadCount * 1000).toISOString(),
+      message: `${String(item?.message ?? "warning")} occurrence=${eventReadCount + 1}`,
+    }))
+    : sourceItems;
+  if (fixture.trackEventReads === true || fixture.raceAfterEvents === true || Array.isArray(fixture.eventsAfterFirstRead)) {
+    state.push({ resource: "read", name: "events", sequence: eventReadCount + 1, valuesPrinted: false });
+    await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`);
+  }
+  console.log(JSON.stringify({ apiVersion: "v1", kind: "List", items }));
   process.exit(0);
 }
 
@@ -54,6 +75,10 @@ if (args[0] === "delete") {
   const selectorIndex = args.indexOf("-l");
   const selector = selectorIndex >= 0 ? String(args[selectorIndex + 1] ?? "") : null;
   const name = directName ?? selector ?? "selected";
+  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, valuesPrinted: false });
   await Bun.write(statePath, `${JSON.stringify(state, null, 2)}\n`);
   console.log(`${resource}/${name.replace(/[^a-z0-9.-]+/giu, "-")} deleted`);