fix: 周期回收陈旧 Pending runner

src/mgr/postgres-store.ts

@@ -906,8 +906,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       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 facts = { run, command, commands, events: eventRows.rows.map(eventFromRow) };
+      if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) return await operation(facts);
 
       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]);
@@ -921,7 +921,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       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;
+      return await operation(facts);
     });
   }
 

src/mgr/runner-reconciler.ts

@@ -6,18 +6,24 @@ import { emitAgentRunOtelSpan } from "../common/otel-trace.js";
 import { nowIso, stableHash } from "../common/validation.js";
 import type { AgentRunStore } from "./store.js";
 import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
+import { reconcileStalePendingRunners, type RunnerRetentionOptions } from "./runner-retention.js";
 
 export interface RunnerReconcilerOptions {
   store: AgentRunStore;
   namespace?: string;
   kubectlCommand?: string;
   limit?: number;
+  intervalMs?: number;
+  nextReconcileAt?: string | null;
+  retention?: RunnerRetentionOptions;
 }
 
 export interface RunnerReconcilerLoopOptions extends RunnerReconcilerOptions {
   batchSize: number;
   intervalMs: number;
   onError?: (error: unknown) => void;
+  onResult?: (result: JsonRecord) => void;
+  onSchedule?: (nextReconcileAt: string | null) => void;
 }
 
 interface K8sObject {
@@ -68,7 +74,9 @@ interface TerminalOutboxSearch {
 }
 
 export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): Promise<JsonRecord> {
+  const startedAt = nowIso();
   const limit = clampLimit(input.limit ?? 20);
+  const stalePendingRetention = await reconcileStalePendingWithinRunnerLoop(input, limit);
   const jobs = await input.store.listRunnerJobsForReconciliation(limit);
   const items: JsonRecord[] = [];
   let observeFailedCount = 0;
@@ -97,19 +105,76 @@ export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): P
     });
   }
 
+  const reconciledAt = nowIso();
+  const nextReconcileAt = input.nextReconcileAt !== undefined
+    ? input.nextReconcileAt
+    : input.intervalMs && input.intervalMs > 0 ? new Date(Date.parse(reconciledAt) + input.intervalMs).toISOString() : null;
   return {
     action: "runner-job-reconcile",
-    reconciledAt: nowIso(),
+    startedAt,
+    reconciledAt,
+    nextReconcileAt,
     limit,
     scannedCount: jobs.length,
     updatedCount: jobs.length,
     observeFailedCount,
     runClosureCount,
+    stalePendingRetention,
     items,
     valuesPrinted: false,
   };
 }
 
+async function reconcileStalePendingWithinRunnerLoop(input: RunnerReconcilerOptions, limit: number): Promise<JsonRecord> {
+  if (!input.retention) {
+    return {
+      enabled: false,
+      action: "runner-stale-pending-reconcile",
+      namespace: input.namespace ?? null,
+      reason: "runner-retention-disabled",
+      valuesPrinted: false,
+    };
+  }
+  const options: RunnerRetentionOptions = {
+    ...input.retention,
+    ...(input.namespace ? { namespace: input.namespace } : {}),
+    ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}),
+  };
+  try {
+    return await reconcileStalePendingRunners({ store: input.store, options, limit });
+  } catch (error) {
+    return {
+      enabled: true,
+      action: "runner-stale-pending-reconcile",
+      namespace: options.namespace,
+      stalePendingMaxAgeMs: options.stalePendingMaxAgeMs,
+      state: "failed-closed",
+      reason: "runner-stale-pending-observation-failed",
+      error: reconcilerErrorSummary(error),
+      valuesPrinted: false,
+    };
+  }
+}
+
+function reconcilerErrorSummary(error: unknown): JsonRecord {
+  if (error instanceof AgentRunError) {
+    return redactJson({
+      failureKind: error.failureKind,
+      httpStatus: error.httpStatus,
+      message: redactText(error.message).slice(0, 300),
+      details: error.details,
+      valuesPrinted: false,
+    }) as JsonRecord;
+  }
+  return {
+    failureKind: "infra-failed",
+    httpStatus: 500,
+    message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 300),
+    details: null,
+    valuesPrinted: false,
+  };
+}
+
 async function recoverTerminalOutboxIfNeeded(store: AgentRunStore, job: RunnerJobRecord, observation: JsonRecord, input: { namespace?: string; kubectlCommand?: string }): Promise<JsonRecord | null> {
   const phase = stringValue(observation.observedRunnerPhase);
   if (phase !== "k8s:succeeded" && phase !== "k8s:failed") return null;
@@ -273,23 +338,37 @@ export function startRunnerJobReconciler(input: RunnerReconcilerLoopOptions): ()
   const intervalMs = Math.max(1_000, input.intervalMs);
   let stopped = false;
   let running = false;
+  let timer: NodeJS.Timeout | null = null;
   const tick = async (): Promise<void> => {
     if (stopped || running) return;
     running = true;
+    input.onSchedule?.(null);
+    let result: JsonRecord | null = null;
     try {
-      await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) });
+      result = await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}), ...(input.retention ? { retention: input.retention } : {}) });
     } catch (error) {
       input.onError?.(error);
     } finally {
       running = false;
     }
+    if (stopped) {
+      input.onSchedule?.(null);
+      return;
+    }
+    const nextReconcileAt = new Date(Date.now() + intervalMs).toISOString();
+    input.onSchedule?.(nextReconcileAt);
+    if (result) {
+      result.nextReconcileAt = nextReconcileAt;
+      input.onResult?.(result);
+    }
+    timer = setTimeout(() => { void tick(); }, intervalMs);
+    timer.unref?.();
   };
-  const timer = setInterval(() => { void tick(); }, intervalMs);
-  timer.unref?.();
   void tick();
   return () => {
     stopped = true;
-    clearInterval(timer);
+    if (timer) clearTimeout(timer);
+    input.onSchedule?.(null);
   };
 }
 

src/mgr/runner-retention.ts

@@ -2,7 +2,7 @@ import { spawn } from "node:child_process";
 import { AgentRunError } from "../common/errors.js";
 import { redactJson, redactText } from "../common/redaction.js";
 import type { CommandRecord, JsonRecord, JsonValue, RunRecord } from "../common/types.js";
-import { stableHash } from "../common/validation.js";
+import { nowIso, stableHash } from "../common/validation.js";
 import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
 import type { AgentRunStore } from "./store.js";
 import { throwIfAborted, waitForChildProcess } from "./abortable-child.js";
@@ -46,6 +46,30 @@ export interface RunnerRetentionSummary extends JsonRecord {
   valuesPrinted: false;
 }
 
+export interface RunnerStalePendingReconcileSummary extends JsonRecord {
+  enabled: true;
+  action: "runner-stale-pending-reconcile";
+  observedAt: string;
+  namespace: string;
+  limit: number;
+  maxRunners: number;
+  stalePendingMaxAgeMs: number;
+  observedRunnerCount: number;
+  liveRunnerCountBefore: number;
+  liveRunnerCountAfter: number | null;
+  stalePendingCandidateCount: number;
+  protectedActiveRunnerCount: number;
+  selectedRunnerCount: number;
+  terminalizedRunCount: number;
+  terminalizedCommandCount: number;
+  deletedRunnerJobCount: number;
+  deletedRunnerPodCount: number;
+  deletedAssociatedResourceCount: number;
+  failedRunnerCount: number;
+  reason: string;
+  valuesPrinted: false;
+}
+
 type CandidateKind = "idle" | "terminal" | "inactive" | "stale-pending";
 
 interface RunnerPodObservation extends JsonRecord {
@@ -165,29 +189,10 @@ 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.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") {
-      deletedRunnerJobCount++;
-      deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
-    } else {
-      deletedRunnerPodCount++;
-    }
+    const retired = await retireSelectedRunner(input.store, input.options, item, `runner retention retired stale Kubernetes ${item.resourceKind}/${item.name}`);
+    deletedRunnerJobCount += retired.deletedRunnerJobCount;
+    deletedRunnerPodCount += retired.deletedRunnerPodCount;
+    if (item.resourceKind === "job") deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
   }
   const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before;
   if (after.liveRunnerCount > targetBeforeCreate) {
@@ -239,6 +244,100 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
   } satisfies RunnerRetentionSummary;
 }
 
+export async function reconcileStalePendingRunners(input: { store: AgentRunStore; options: RunnerRetentionOptions; limit: number }): Promise<RunnerStalePendingReconcileSummary> {
+  const limit = clampReconcileLimit(input.limit);
+  return await input.store.withRunnerCreateFence(input.options.namespace, async () => {
+    const observedAt = nowIso();
+    const before = await runnerSnapshot(input.store, input.options);
+    const resources = [...before.jobs, ...before.pods];
+    const stalePendingCandidates = resources
+      .filter((item) => !item.protectedActive && item.candidateKind === "stale-pending")
+      .sort(compareCandidates);
+    const protectedActiveRunners = resources.filter((item) => item.protectedActive);
+    const selected = stalePendingCandidates.slice(0, limit);
+    const outcomes: JsonRecord[] = [];
+    let terminalizedRunCount = 0;
+    let terminalizedCommandCount = 0;
+    let deletedRunnerJobCount = 0;
+    let deletedRunnerPodCount = 0;
+    let deletedAssociatedResourceCount = 0;
+    let failedRunnerCount = 0;
+
+    for (const item of selected) {
+      let retired: RetiredRunnerResource | null = null;
+      try {
+        retired = await retireSelectedRunner(input.store, input.options, item, `runner reconciler retired stale-pending Kubernetes ${item.resourceKind}/${item.name}`);
+        terminalizedRunCount += item.candidateKind === "stale-pending" ? 1 : 0;
+        terminalizedCommandCount += item.candidateKind === "stale-pending" ? 1 : 0;
+        deletedRunnerJobCount += retired.deletedRunnerJobCount;
+        deletedRunnerPodCount += retired.deletedRunnerPodCount;
+        const deletedAssociatedForRunner = item.resourceKind === "job" ? await deleteAssociatedResources(input.options, item.jobName) : 0;
+        deletedAssociatedResourceCount += deletedAssociatedForRunner;
+        outcomes.push({ ...resourceIdentitySummary(item), state: "retired", terminalized: item.candidateKind === "stale-pending", ...retired, deletedAssociatedResourceCount: deletedAssociatedForRunner, valuesPrinted: false });
+      } catch (error) {
+        failedRunnerCount++;
+        outcomes.push({ ...resourceIdentitySummary(item), state: retired ? "retired-associated-cleanup-failed" : "failed-closed", coreRetired: retired !== null, ...(retired ?? {}), error: retentionErrorSummary(error), valuesPrinted: false });
+      }
+    }
+
+    let after: Snapshot | null = before;
+    let postObservationError: JsonRecord | null = null;
+    if (selected.length > 0) {
+      try {
+        after = await runnerSnapshot(input.store, input.options);
+      } catch (error) {
+        after = null;
+        postObservationError = retentionErrorSummary(error);
+      }
+    }
+    const candidateSummaries = boundedResourceSummaries(stalePendingCandidates);
+    const protectedSummaries = boundedResourceSummaries(protectedActiveRunners);
+    const selectedSummaries = boundedResourceSummaries(selected);
+    const boundedOutcomes = boundedJsonRecords(outcomes);
+    return {
+      enabled: true,
+      action: "runner-stale-pending-reconcile",
+      observedAt,
+      namespace: input.options.namespace,
+      limit,
+      maxRunners: input.options.maxRunners,
+      stalePendingMaxAgeMs: input.options.stalePendingMaxAgeMs,
+      observedRunnerCount: resources.length,
+      liveRunnerCountBefore: before.liveRunnerCount,
+      liveRunnerCountAfter: after?.liveRunnerCount ?? null,
+      runnerJobCountBefore: before.runnerJobCount,
+      nonTerminalRunnerPodCountBefore: before.nonTerminalRunnerPodCount,
+      orphanNonTerminalRunnerPodCountBefore: before.orphanNonTerminalRunnerPodCount,
+      inactiveCandidateCount: resources.filter((item) => !item.protectedActive).length,
+      stalePendingCandidateCount: stalePendingCandidates.length,
+      stalePendingCandidates: candidateSummaries.items,
+      stalePendingCandidateOmittedCount: candidateSummaries.omittedCount,
+      protectedActiveRunnerCount: protectedActiveRunners.length,
+      protectedActiveRunners: protectedSummaries.items,
+      protectedActiveRunnerOmittedCount: protectedSummaries.omittedCount,
+      protectedReasonCounts: reasonCounts(protectedActiveRunners.map((item) => item.protectedReason ?? "unknown")),
+      selectedRunnerCount: selected.length,
+      selectedRunners: selectedSummaries.items,
+      selectedRunnerOmittedCount: selectedSummaries.omittedCount,
+      terminalizedRunCount,
+      terminalizedCommandCount,
+      deletedRunnerJobCount,
+      deletedRunnerPodCount,
+      deletedAssociatedResourceCount,
+      failedRunnerCount,
+      outcomes: boundedOutcomes.items,
+      outcomeOmittedCount: boundedOutcomes.omittedCount,
+      state: postObservationError || failedRunnerCount > 0 ? "partial" : "completed",
+      postObservationState: postObservationError ? "failed" : "completed",
+      postObservationError,
+      activeRunRisk: protectedActiveRunners.length > 0,
+      reason: postObservationError ? "stale-pending-post-observation-failed" : selected.length === 0 ? "no-stale-pending-candidate" : failedRunnerCount > 0 ? "stale-pending-reconcile-partial" : "stale-pending-reconciled",
+      cleanupPolicy: cleanupPolicySummary(),
+      valuesPrinted: false,
+    } satisfies RunnerStalePendingReconcileSummary;
+  });
+}
+
 async function runnerSnapshot(store: AgentRunStore, options: RunnerRetentionOptions): Promise<Snapshot> {
   const observed = await Promise.allSettled([
     kubectlGetList(options, "jobs", labelSelector(options.matchLabels)),
@@ -419,6 +518,35 @@ async function activityFor(store: AgentRunStore, input: {
   return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint);
 }
 
+interface RetiredRunnerResource extends JsonRecord {
+  deletedRunnerJobCount: number;
+  deletedRunnerPodCount: number;
+}
+
+async function retireSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, item: RunnerResourceEntry, reason: string): Promise<RetiredRunnerResource> {
+  await revalidateSelectedRunner(store, 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: options.namespace, resourceKind: item.resourceKind, name: item.name, valuesPrinted: false },
+    });
+  }
+  await store.withRunnerRetentionFence({
+    runId: item.runId,
+    commandId: item.commandId,
+    terminalizeStalePending: item.candidateKind === "stale-pending",
+    reason,
+  }, async (facts) => {
+    const currentDbFingerprint = databaseFactsFingerprint(facts.run, facts.command, facts.commands, facts.events);
+    if (currentDbFingerprint !== item.dbFactsFingerprint) throw databaseFenceRejected(options, item, currentDbFingerprint, facts.run);
+    await deleteRunnerResourceCas(options, item);
+  });
+  if (item.resourceKind === "job") {
+    return { deletedRunnerJobCount: 1, deletedRunnerPodCount: 0 };
+  }
+  return { deletedRunnerJobCount: 0, deletedRunnerPodCount: 1 };
+}
+
 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;
@@ -599,13 +727,22 @@ function resourceSummary(item: RunnerResourceEntry): JsonRecord {
     classificationReason: item.classificationReason,
     terminal: item.terminal,
     podPhase: item.podPhase,
-    podObservations: item.podObservations.slice(0, 5),
+    podObservations: item.podObservations.slice(0, 5).map(podObservationSummary),
     podObservationOmittedCount: Math.max(0, item.podObservations.length - 5),
     lastActiveAt: new Date(item.lastActiveAtMs).toISOString(),
     valuesPrinted: false,
   }) as JsonRecord;
 }
 
+function podObservationSummary(item: RunnerPodObservation): JsonRecord {
+  return {
+    ...item,
+    waiting: item.waiting.slice(0, 5),
+    waitingOmittedCount: Math.max(0, item.waiting.length - 5),
+    valuesPrinted: false,
+  };
+}
+
 function cleanupPolicySummary(): JsonRecord {
   return {
     mode: "db-ledger-and-k8s-observation",
@@ -874,6 +1011,54 @@ function boundedResourceSummaries(items: RunnerResourceEntry[]): { items: JsonRe
   return { items: items.slice(0, limit).map(resourceSummary), omittedCount: Math.max(0, items.length - limit) };
 }
 
+function boundedJsonRecords(items: JsonRecord[]): { items: JsonRecord[]; omittedCount: number } {
+  const limit = 20;
+  return { items: items.slice(0, limit), omittedCount: Math.max(0, items.length - limit) };
+}
+
+function resourceIdentitySummary(item: RunnerResourceEntry): JsonRecord {
+  return {
+    resourceKind: item.resourceKind,
+    name: item.name,
+    jobName: item.jobName,
+    runId: item.runId,
+    commandId: item.commandId,
+    runnerId: item.runnerId,
+    candidateKind: item.candidateKind,
+    classificationReason: item.classificationReason,
+    valuesPrinted: false,
+  };
+}
+
+function retentionErrorSummary(error: unknown): JsonRecord {
+  if (error instanceof AgentRunError) {
+    return redactJson({
+      failureKind: error.failureKind,
+      httpStatus: error.httpStatus,
+      message: redactText(error.message).slice(0, 300),
+      details: error.details,
+      valuesPrinted: false,
+    }) as JsonRecord;
+  }
+  return {
+    failureKind: "infra-failed",
+    httpStatus: 500,
+    message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 300),
+    details: null,
+    valuesPrinted: false,
+  };
+}
+
+function reasonCounts(reasons: string[]): JsonRecord {
+  const counts: Record<string, number> = {};
+  for (const reason of reasons) counts[reason] = (counts[reason] ?? 0) + 1;
+  return Object.fromEntries(Object.entries(counts).sort(([left], [right]) => left.localeCompare(right))) as JsonRecord;
+}
+
+function clampReconcileLimit(limit: number): number {
+  return Math.max(1, Math.min(Math.floor(limit), 500));
+}
+
 function jsonRecord(value: unknown): JsonRecord | null {
   return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : null;
 }

src/mgr/server.ts

@@ -115,13 +115,42 @@ function runnerRetentionOptionsForRequest(defaults: RunnerRetentionOptions | und
   };
 }
 
-function runnerReconcilerOptionsForRuntime(defaults: ManagerServerOptions["runnerReconcilerOptions"] | undefined, runnerJobDefaults: ManagerServerOptions["runnerJobDefaults"] | undefined): RunnerReconcilerRuntimeOptions {
+function runnerReconcilerOptionsForRuntime(defaults: ManagerServerOptions["runnerReconcilerOptions"] | undefined, runnerJobDefaults: RunnerJobDefaults): RunnerReconcilerRuntimeOptions {
   const enabled = defaults?.enabled ?? optionalBooleanEnv("AGENTRUN_MANAGER_RECONCILER_ENABLED", process.env.AGENTRUN_MANAGER_RECONCILER_ENABLED, false);
   const namespace = defaults?.namespace ?? runnerJobDefaults?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE ?? "agentrun-v01";
   const intervalMs = defaults?.intervalMs ?? optionalPositiveInteger("AGENTRUN_MANAGER_RECONCILER_INTERVAL_MS", process.env.AGENTRUN_MANAGER_RECONCILER_INTERVAL_MS) ?? 30_000;
   const batchSize = defaults?.batchSize ?? optionalPositiveInteger("AGENTRUN_MANAGER_RECONCILER_BATCH_SIZE", process.env.AGENTRUN_MANAGER_RECONCILER_BATCH_SIZE) ?? 20;
   const kubectlCommand = defaults?.kubectlCommand ?? runnerJobDefaults?.kubectlCommand;
-  return { enabled, namespace, intervalMs, batchSize, ...(kubectlCommand ? { kubectlCommand } : {}) };
+  return { enabled, namespace, intervalMs, batchSize, nextReconcileAt: null, ...(kubectlCommand ? { kubectlCommand } : {}), ...(runnerJobDefaults.retention ? { retention: runnerJobDefaults.retention } : {}) };
+}
+
+function runnerReconcilerLogSummary(result: JsonRecord): JsonRecord {
+  const retention = typeof result.stalePendingRetention === "object" && result.stalePendingRetention !== null && !Array.isArray(result.stalePendingRetention)
+    ? result.stalePendingRetention as JsonRecord
+    : {};
+  return {
+    code: "runner-reconciler-completed",
+    reconciledAt: result.reconciledAt ?? null,
+    nextReconcileAt: result.nextReconcileAt ?? null,
+    scannedCount: result.scannedCount ?? 0,
+    observeFailedCount: result.observeFailedCount ?? 0,
+    runClosureCount: result.runClosureCount ?? 0,
+    stalePendingRetention: {
+      enabled: retention.enabled === true,
+      state: retention.state ?? "completed",
+      reason: retention.reason ?? null,
+      observedRunnerCount: retention.observedRunnerCount ?? 0,
+      stalePendingCandidateCount: retention.stalePendingCandidateCount ?? 0,
+      protectedActiveRunnerCount: retention.protectedActiveRunnerCount ?? 0,
+      protectedReasonCounts: retention.protectedReasonCounts ?? {},
+      selectedRunnerCount: retention.selectedRunnerCount ?? 0,
+      terminalizedRunCount: retention.terminalizedRunCount ?? 0,
+      deletedRunnerJobCount: retention.deletedRunnerJobCount ?? 0,
+      failedRunnerCount: retention.failedRunnerCount ?? 0,
+      valuesPrinted: false,
+    },
+    valuesPrinted: false,
+  };
 }
 
 function runnerDispatcherOptionsForRuntime(defaults: ManagerServerOptions["runnerDispatcherOptions"] | undefined): RunnerDispatcherOptions {
@@ -201,7 +230,9 @@ interface RunnerReconcilerRuntimeOptions {
   namespace: string;
   intervalMs: number;
   batchSize: number;
+  nextReconcileAt: string | null;
   kubectlCommand?: string;
+  retention?: RunnerRetentionOptions;
 }
 
 export interface StartedManagerServer {
@@ -217,15 +248,25 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr
   const sessionPvcDefaults = options.sessionPvcOptions;
   const providerProfileDefaults = options.providerProfileOptions;
   const toolCredentialDefaults = options.toolCredentialOptions;
-  const runnerReconcilerOptions = runnerReconcilerOptionsForRuntime(options.runnerReconcilerOptions, runnerJobDefaults);
+  const resolvedRunnerJobDefaults = runnerJobDefaultsForRequest(runnerJobDefaults, sourceCommit);
+  const runnerReconcilerOptions = runnerReconcilerOptionsForRuntime(options.runnerReconcilerOptions, resolvedRunnerJobDefaults);
   const runnerDispatcherOptions = runnerDispatcherOptionsForRuntime(options.runnerDispatcherOptions);
   const kafkaOutboxRelayOptions = kafkaOutboxRelayOptionsForRuntime(options.kafkaOutboxRelayOptions);
-  const resolvedRunnerJobDefaults = runnerJobDefaultsForRequest(runnerJobDefaults, sourceCommit);
   assertRunnerRuntimeConfigured(runnerJobDefaults);
   const aipodSpecDir = options.aipodSpecDir ?? process.env.AGENTRUN_AIPOD_SPEC_DIR;
   const auth = options.auth ?? managerServerAuthConfigFromEnv();
   const authSummary = managerAuthSummary(auth);
-  const stopRunnerReconciler = runnerReconcilerOptions.enabled ? startRunnerJobReconciler({ store, namespace: runnerReconcilerOptions.namespace, intervalMs: runnerReconcilerOptions.intervalMs, batchSize: runnerReconcilerOptions.batchSize, ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}), onError: (error) => console.warn(JSON.stringify({ code: "runner-reconciler-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null;
+  const stopRunnerReconciler = runnerReconcilerOptions.enabled ? startRunnerJobReconciler({
+    store,
+    namespace: runnerReconcilerOptions.namespace,
+    intervalMs: runnerReconcilerOptions.intervalMs,
+    batchSize: runnerReconcilerOptions.batchSize,
+    ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}),
+    ...(runnerReconcilerOptions.retention ? { retention: runnerReconcilerOptions.retention } : {}),
+    onSchedule: (nextReconcileAt) => { runnerReconcilerOptions.nextReconcileAt = nextReconcileAt; },
+    onResult: (result) => console.info(JSON.stringify(runnerReconcilerLogSummary(result))),
+    onError: (error) => console.warn(JSON.stringify({ code: "runner-reconciler-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })),
+  }) : null;
   const stopRunnerDispatcher = runnerDispatcherOptions.enabled ? startRunnerDispatcher({ store, defaults: resolvedRunnerJobDefaults, options: runnerDispatcherOptions, onError: (error) => console.warn(JSON.stringify({ code: "runner-dispatcher-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null;
   const stopKafkaOutboxRelay = kafkaOutboxRelayOptions.enabled ? startKafkaEventOutboxRelay({ store, options: kafkaOutboxRelayOptions, onError: (error) => console.warn(JSON.stringify({ code: "kafka-outbox-relay-failed", message: error instanceof Error ? error.message : String(error), valuesPrinted: false })) }) : null;
   const server = createServer(async (req, res) => {
@@ -869,7 +910,14 @@ async function route({ method, url, body, store, sourceCommit, authSummary, runn
     const record = body === null ? {} : asRecord(body, "runnerReconciler");
     const limit = Math.max(1, Math.min(numberField(record, "limit", runnerReconcilerOptions?.batchSize ?? 20), 500));
     const namespace = optionalString(record.namespace) ?? runnerReconcilerOptions?.namespace ?? process.env.AGENTRUN_RUNTIME_NAMESPACE ?? "agentrun-v01";
-    return await reconcileRunnerJobsOnce({ store, namespace, limit, ...(runnerReconcilerOptions?.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}) }) as JsonValue;
+    return await reconcileRunnerJobsOnce({
+      store,
+      namespace,
+      limit,
+      nextReconcileAt: runnerReconcilerOptions.nextReconcileAt,
+      ...(runnerReconcilerOptions.kubectlCommand ? { kubectlCommand: runnerReconcilerOptions.kubectlCommand } : {}),
+      ...(runnerReconcilerOptions.retention ? { retention: runnerReconcilerOptions.retention } : {}),
+    }) as JsonValue;
   }
   if (method === "POST" && path === "/api/v1/reconciler/runner-dispatch") {
     return await dispatchRunnerIntentsOnce({ store, defaults: resolvedRunnerJobDefaults, options: runnerDispatcherOptions }) as JsonValue;

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

@@ -3,7 +3,10 @@ import { chmod, readFile, writeFile } from "node:fs/promises";
 import path from "node:path";
 import { AgentRunError } from "../../common/errors.js";
 import type { CommandRecord, JsonRecord, RunEvent, RunRecord } from "../../common/types.js";
+import { ManagerClient } from "../../mgr/client.js";
+import { reconcileRunnerJobsOnce, startRunnerJobReconciler } from "../../mgr/runner-reconciler.js";
 import { enforceRunnerRetentionBeforeCreate, type RunnerRetentionOptions, type RunnerRetentionSummary } from "../../mgr/runner-retention.js";
+import { startManagerServer } from "../../mgr/server.js";
 import { MemoryAgentRunStore, type RunnerRetentionFenceFacts, type RunnerRetentionFenceInput } from "../../mgr/store.js";
 import type { SelfTestCase } from "../harness.js";
 
@@ -41,6 +44,7 @@ interface RetentionFixture {
 class RetentionFactStore extends MemoryAgentRunStore {
   private readonly facts: RetentionFact[];
   private beforeFence: (() => void) | null = null;
+  terminalizationCount = 0;
 
   constructor(facts: RetentionFact[]) {
     super();
@@ -90,6 +94,7 @@ class RetentionFactStore extends MemoryAgentRunStore {
     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) {
+      this.terminalizationCount++;
       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 };
@@ -119,6 +124,12 @@ const selfTest: SelfTestCase = async (context) => {
     await assertDeletionPostconditionFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertMemoryDbFenceRejectsConcurrentClaim();
     await assertRunnerCreateFenceSerializesNamespace();
+    await assertPeriodicReconcilerCleansAtCapacity({ fixturePath, statePath, kubectlCommand });
+    await assertPeriodicReconcilerProtectsUnsafeRunners({ fixturePath, statePath, kubectlCommand });
+    await assertPeriodicReconcilerCasConflictFailsClosed({ fixturePath, statePath, kubectlCommand });
+    await assertConcurrentPeriodicReconcileIsSerialized({ fixturePath, statePath, kubectlCommand });
+    await assertControlledReconcileApiExposesRetention({ fixturePath, statePath, kubectlCommand });
+    await assertAutomaticReconcilerReportsNextSchedule();
   } finally {
     restoreEnv("AGENTRUN_SELFTEST_RETENTION_FIXTURE_PATH", previousFixturePath);
     restoreEnv("AGENTRUN_SELFTEST_RETENTION_STATE_PATH", previousStatePath);
@@ -149,6 +160,12 @@ const selfTest: SelfTestCase = async (context) => {
       "runner-delete-postcondition-checked",
       "memory-db-fence-rejects-concurrent-claim",
       "namespace-runner-create-fence-prevents-capacity-overshoot",
+      "periodic-reconciler-cleans-stale-pending-at-capacity",
+      "periodic-reconciler-protects-unsafe-runners",
+      "periodic-reconciler-cas-conflict-fails-closed",
+      "periodic-reconciler-namespace-fence-prevents-duplicate-terminalization",
+      "controlled-reconcile-api-exposes-bounded-retention-result",
+      "automatic-reconciler-reports-next-schedule",
     ],
   };
 };
@@ -420,6 +437,226 @@ async function assertRunnerCreateFenceSerializesNamespace(): Promise<void> {
   assert.equal(liveRunnerCount, maxRunners);
 }
 
+async function assertPeriodicReconcilerCleansAtCapacity(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const now = Date.now();
+  const facts = Array.from({ length: 20 }, (_, index) => pendingFact(index + 300, new Date(now - (60 - index) * 60_000).toISOString()));
+  await writeFixture(input, {
+    jobs: facts.map((fact, index) => runnerJob(fact, index + 300)),
+    pods: facts.map((fact, index) => runnerPod(fact, index + 300, "pending")),
+    events: facts.map((fact, index) => failedMountEvent(fact, index + 300)),
+    bumpJobResourceVersionAfterPodDelete: true,
+    bumpedJobResourceVersion: "2",
+  });
+  const store = new RetentionFactStore(facts);
+  const summary = await reconcileRunnerJobsOnce({
+    store,
+    namespace: "agentrun-nc01-v02",
+    kubectlCommand: input.kubectlCommand,
+    retention: retentionOptions(input.kubectlCommand, 20),
+    limit: 3,
+    intervalMs: 30_000,
+  });
+  const retention = summary.stalePendingRetention as JsonRecord;
+  assert.equal(retention.liveRunnerCountBefore, 20);
+  assert.equal(retention.liveRunnerCountAfter, 17);
+  assert.equal(retention.stalePendingCandidateCount, 20);
+  assert.equal(retention.selectedRunnerCount, 3);
+  assert.equal(retention.terminalizedRunCount, 3);
+  assert.equal(retention.terminalizedCommandCount, 3);
+  assert.equal(retention.deletedRunnerJobCount, 3);
+  assert.equal(retention.failedRunnerCount, 0);
+  assert.equal(retention.reason, "stale-pending-reconciled");
+  assert.equal(store.terminalizationCount, 3);
+  for (const fact of facts.slice(0, 3)) {
+    assert.equal(store.getRun(fact.run.id).status, "failed");
+    assert.equal(store.getCommand(fact.command.id).state, "failed");
+  }
+  assert.equal(store.getRun((facts[3] as RetentionFact).run.id).status, "pending");
+  const reconciledAt = Date.parse(String(summary.reconciledAt));
+  const nextReconcileAt = Date.parse(String(summary.nextReconcileAt));
+  assert.equal(nextReconcileAt - reconciledAt, 30_000);
+  assert.deepEqual(
+    deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name),
+    ["agentrun-v02-runner-stale-300", "agentrun-v02-runner-stale-301", "agentrun-v02-runner-stale-302"],
+  );
+}
+
+async function assertPeriodicReconcilerProtectsUnsafeRunners(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const oldAt = new Date(Date.now() - 60 * 60_000).toISOString();
+  const freshAt = new Date(Date.now() - 60_000).toISOString();
+  const fresh = pendingFact(320, freshAt);
+  const running = pendingFact(321, oldAt);
+  const ready = pendingFact(322, oldAt);
+  const claimed = pendingFact(323, oldAt, "claimed");
+  const toolExecuting = pendingFact(324, oldAt, "tool-executing");
+  const unavailable = pendingFact(325, oldAt);
+  const allFacts = [fresh, running, ready, claimed, toolExecuting, unavailable];
+  await writeFixture(input, {
+    jobs: allFacts.map((fact, index) => runnerJob(fact, index + 320)),
+    pods: [
+      runnerPod(fresh, 320, "pending"),
+      runnerPod(running, 321, "running"),
+      runnerPod(ready, 322, "ready"),
+      runnerPod(claimed, 323, "pending"),
+      runnerPod(toolExecuting, 324, "failed-config"),
+      runnerPod(unavailable, 325, "pending"),
+    ],
+    events: allFacts.map((fact, index) => failedMountEvent(fact, index + 320)),
+  });
+  const summary = await reconcileRunnerJobsOnce({
+    store: new RetentionFactStore([fresh, running, ready, claimed, toolExecuting]),
+    namespace: "agentrun-nc01-v02",
+    kubectlCommand: input.kubectlCommand,
+    retention: retentionOptions(input.kubectlCommand, 20),
+    limit: 20,
+  });
+  const retention = summary.stalePendingRetention as JsonRecord;
+  assert.equal(retention.stalePendingCandidateCount, 0);
+  assert.equal(retention.selectedRunnerCount, 0);
+  assert.equal(retention.protectedActiveRunnerCount, 6);
+  const reasonCounts = retention.protectedReasonCounts as JsonRecord;
+  assert.equal(reasonCounts["stale-pending-window-active"], 1);
+  assert.equal(reasonCounts["runner-pod-started-running-or-ready"], 2);
+  assert.equal(reasonCounts["fresh-active-run"], 1);
+  assert.equal(reasonCounts["business-progress-observed-or-unverifiable"], 1);
+  assert.equal(reasonCounts["db-facts-unavailable"], 1);
+  assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
+}
+
+async function assertPeriodicReconcilerCasConflictFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(330, new Date(Date.now() - 60 * 60_000).toISOString());
+  const racedPod = runnerPod(fact, 330, "running");
+  (racedPod.metadata as JsonRecord).resourceVersion = "2";
+  await writeFixture(input, {
+    jobs: [runnerJob(fact, 330)],
+    pods: [runnerPod(fact, 330, "pending")],
+    events: [failedMountEvent(fact, 330)],
+    raceOnCasDelete: true,
+    racePods: [racedPod],
+  });
+  const store = new RetentionFactStore([fact]);
+  const summary = await reconcileRunnerJobsOnce({
+    store,
+    namespace: "agentrun-nc01-v02",
+    kubectlCommand: input.kubectlCommand,
+    retention: retentionOptions(input.kubectlCommand, 20),
+    limit: 1,
+  });
+  const retention = summary.stalePendingRetention as JsonRecord;
+  assert.equal(retention.selectedRunnerCount, 1);
+  assert.equal(retention.failedRunnerCount, 1);
+  assert.equal(retention.terminalizedRunCount, 0);
+  assert.equal(retention.deletedRunnerJobCount, 0);
+  assert.equal(store.terminalizationCount, 0);
+  assert.equal(store.getRun(fact.run.id).status, "pending");
+  assert.equal(store.getCommand(fact.command.id).state, "pending");
+  const outcome = ((retention.outcomes as JsonRecord[])[0] ?? {}) as JsonRecord;
+  assert.equal(outcome.state, "failed-closed");
+  const error = (outcome.error ?? {}) as JsonRecord;
+  const details = (error.details ?? {}) as JsonRecord;
+  assert.equal(details.reason, "runner-retention-k8s-cas-rejected");
+  assert.equal(details.changeReason, "object-identity-or-version-changed");
+  assert.equal(deletionOperations(await readDeletionState(input.statePath)).length, 0);
+}
+
+async function assertConcurrentPeriodicReconcileIsSerialized(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(340, new Date(Date.now() - 60 * 60_000).toISOString());
+  await writeFixture(input, {
+    jobs: [runnerJob(fact, 340)],
+    pods: [runnerPod(fact, 340, "pending")],
+    events: [failedMountEvent(fact, 340)],
+    bumpJobResourceVersionAfterPodDelete: true,
+    bumpedJobResourceVersion: "2",
+  });
+  const store = new RetentionFactStore([fact]);
+  const options = {
+    store,
+    namespace: "agentrun-nc01-v02",
+    kubectlCommand: input.kubectlCommand,
+    retention: retentionOptions(input.kubectlCommand, 20),
+    limit: 1,
+  };
+  const [first, second] = await Promise.all([reconcileRunnerJobsOnce(options), reconcileRunnerJobsOnce(options)]);
+  const summaries = [first.stalePendingRetention as JsonRecord, second.stalePendingRetention as JsonRecord];
+  assert.equal(summaries.reduce((total, item) => total + Number(item.selectedRunnerCount ?? 0), 0), 1);
+  assert.equal(summaries.reduce((total, item) => total + Number(item.terminalizedRunCount ?? 0), 0), 1);
+  assert.equal(store.terminalizationCount, 1);
+  assert.equal(store.getRun(fact.run.id).status, "failed");
+  assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-340"]);
+}
+
+async function assertControlledReconcileApiExposesRetention(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(350, new Date(Date.now() - 60 * 60_000).toISOString());
+  await writeFixture(input, {
+    jobs: [runnerJob(fact, 350)],
+    pods: [runnerPod(fact, 350, "pending")],
+    events: [failedMountEvent(fact, 350)],
+    bumpJobResourceVersionAfterPodDelete: true,
+    bumpedJobResourceVersion: "2",
+  });
+  const store = new RetentionFactStore([fact]);
+  const retention = retentionOptions(input.kubectlCommand, 20);
+  const server = await startManagerServer({
+    port: 0,
+    host: "127.0.0.1",
+    sourceCommit: "self-test-282",
+    store,
+    runnerJobDefaults: {
+      namespace: "agentrun-nc01-v02",
+      managerUrl: "http://agentrun-mgr.agentrun-nc01-v02.svc.cluster.local:8080",
+      runnerApiKeySecretRef: { name: "agentrun-v02-runner-api-key", key: "api-key" },
+      image: "registry.example.invalid/agentrun@sha256:1111111111111111111111111111111111111111111111111111111111111111",
+      serviceAccountName: "agentrun-v02-runner",
+      jobNamePrefix: "agentrun-v02-runner",
+      lane: "v0.2",
+      kubectlCommand: input.kubectlCommand,
+      retention,
+    },
+    runnerReconcilerOptions: {
+      enabled: false,
+      namespace: "agentrun-nc01-v02",
+      kubectlCommand: input.kubectlCommand,
+      intervalMs: 45_000,
+      batchSize: 1,
+    },
+  });
+  try {
+    const summary = await new ManagerClient(server.baseUrl).post("/api/v1/reconciler/runner-jobs", { limit: 1, namespace: "agentrun-nc01-v02" }) as JsonRecord;
+    const cleanup = summary.stalePendingRetention as JsonRecord;
+    assert.equal(cleanup.enabled, true);
+    assert.equal(cleanup.selectedRunnerCount, 1);
+    assert.equal(cleanup.terminalizedRunCount, 1);
+    assert.equal(cleanup.deletedRunnerJobCount, 1);
+    assert.equal(cleanup.failedRunnerCount, 0);
+    assert.equal(cleanup.valuesPrinted, false);
+    assert.equal(summary.nextReconcileAt, null);
+    assert.equal(JSON.stringify(summary).includes("abc123-secret-value"), false);
+  } finally {
+    await new Promise<void>((resolve) => server.server.close(() => resolve()));
+  }
+}
+
+async function assertAutomaticReconcilerReportsNextSchedule(): Promise<void> {
+  let scheduledAt: string | null = null;
+  let resolveResult = (_result: JsonRecord): void => {};
+  const resultPromise = new Promise<JsonRecord>((resolve) => { resolveResult = resolve; });
+  const stop = startRunnerJobReconciler({
+    store: new MemoryAgentRunStore(),
+    batchSize: 1,
+    intervalMs: 1_000,
+    onSchedule: (nextReconcileAt) => { scheduledAt = nextReconcileAt; },
+    onResult: resolveResult,
+  });
+  try {
+    const result = await resultPromise;
+    assert.equal(typeof result.nextReconcileAt, "string");
+    assert.equal(result.nextReconcileAt, scheduledAt);
+    assert.ok(Date.parse(String(result.nextReconcileAt)) > Date.parse(String(result.reconciledAt)));
+  } finally {
+    stop();
+  }
+}
+
 async function assertProtected(
   input: { fixturePath: string; statePath: string; kubectlCommand: string },
   facts: RetentionFact[],

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

@@ -11,7 +11,9 @@ 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"] }));
+  await assertRetentionTerminalizationIsIdempotent();
+  await assertRetentionTerminalizationRollsBackOnCasFailure();
+  console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-run-retention-fence-blocks-claim", "postgres-run-retention-terminalization-idempotent", "postgres-run-retention-cas-failure-rolls-back-terminalization"] }));
 } finally {
   await Promise.all([primary.close(), concurrent.close()]);
 }
@@ -75,6 +77,55 @@ async function assertRetentionFenceBlocksClaim(): Promise<void> {
   assert.equal((await primary.getCommand(command.id)).state, "failed");
 }
 
+async function assertRetentionTerminalizationIsIdempotent(): Promise<void> {
+  const run = await primary.createRun({
+    tenantId: "selftest",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-retention-idempotent" },
+    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 idempotence self-test" } });
+  const fenceInput = { runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "postgres retention idempotence self-test" };
+  await primary.withRunnerRetentionFence(fenceInput, async (facts) => {
+    assert.equal(facts.run.status, "pending");
+  });
+  await primary.withRunnerRetentionFence(fenceInput, async (facts) => {
+    assert.equal(facts.run.status, "failed");
+    assert.equal(facts.command.state, "failed");
+  });
+  const events = await primary.listEvents(run.id, 0, 500);
+  assert.equal(events.filter((event) => event.type === "backend_status" && event.payload.retentionFence === true).length, 1);
+  assert.equal(events.filter((event) => event.type === "terminal_status" && event.payload.retentionFence === true).length, 1);
+  assert.equal((await primary.getRun(run.id)).status, "failed");
+  assert.equal((await primary.getCommand(command.id)).state, "failed");
+}
+
+async function assertRetentionTerminalizationRollsBackOnCasFailure(): Promise<void> {
+  const run = await primary.createRun({
+    tenantId: "selftest",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-retention-rollback" },
+    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 rollback self-test" } });
+  await assert.rejects(
+    primary.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, reason: "postgres retention rollback self-test" }, async () => {
+      throw new AgentRunError("infra-failed", "simulated Kubernetes CAS conflict", { httpStatus: 409, details: { reason: "runner-retention-k8s-cas-rejected", valuesPrinted: false } });
+    }),
+    (error) => error instanceof AgentRunError && error.details?.reason === "runner-retention-k8s-cas-rejected",
+  );
+  assert.equal((await primary.getRun(run.id)).status, "pending");
+  assert.equal((await primary.getCommand(command.id)).state, "pending");
+  const events = await primary.listEvents(run.id, 0, 500);
+  assert.equal(events.filter((event) => event.payload.retentionFence === true).length, 0);
+}
+
 async function delay(ms: number): Promise<void> {
   await new Promise<void>((resolve) => setTimeout(resolve, ms));
 }