Merge pull request #311 from pikasTech/fix/309-terminal-runner-lifecycle

修复终态 runner claim 假活与容量占用

src/mgr/postgres-store.ts

@@ -997,7 +997,10 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       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 facts = { run, command, commands, events: eventRows.rows.map(eventFromRow) };
-      if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) return await operation(facts);
+      if (!input.terminalizeStalePending || isTerminalRunStatus(run.status)) {
+        if (input.releaseRunnerClaim && (isTerminalRunStatus(run.status) || isTerminalCommandState(command.state))) await this.releaseRunnerClaim(client, run);
+        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]);
@@ -1067,7 +1070,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     return this.withTransaction(async (client) => {
       const identity = await client.query("SELECT run_id FROM agentrun_commands WHERE id = $1", [commandId]);
       if (!identity.rows[0]) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
-      const run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id));
+      let run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id));
+      if (isTerminalRunStatus(run.status)) run = await this.releaseTerminalRunClaim(client, run);
       const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
       const row = existing.rows[0];
       if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
@@ -1104,15 +1108,17 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       await client.query("SELECT id FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [runId]);
       if (isTerminalRunStatus(existing.status)) {
         await fenceActiveRunnerDispatchIntents(client, { runId }, "run is terminal");
-        if (existing.status === "cancelled" && result.terminalStatus !== "cancelled") await this.appendEventWithLockedRun(client, runId, "backend_status", lateWriteRejectedPayload(existing, null, { source: "run-status", terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null }));
-        return existing;
+        const released = await this.releaseTerminalRunClaim(client, existing);
+        if (released.status === "cancelled" && result.terminalStatus !== "cancelled") await this.appendEventWithLockedRun(client, runId, "backend_status", lateWriteRejectedPayload(released, null, { source: "run-status", terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null }));
+        return released;
       }
       const status = statusFromTerminal(result.terminalStatus);
       const at = nowIso();
       const updated = await client.query(
-        `UPDATE agentrun_runs SET status = $2, terminal_status = $3, failure_kind = $4, failure_message = $5, updated_at = $6 WHERE id = $1 RETURNING *`,
+        `UPDATE agentrun_runs SET status = $2, terminal_status = $3, failure_kind = $4, failure_message = $5, claimed_by = NULL, lease_expires_at = NULL, updated_at = $6 WHERE id = $1 RETURNING *`,
         [runId, status, result.terminalStatus, result.failureKind, result.failureMessage, at],
       );
+      await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [runId]);
       const run = runFromRow(updated.rows[0]);
       await fenceActiveRunnerDispatchIntents(client, { runId }, `run terminalized as ${status}`, at);
       if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
@@ -1128,7 +1134,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       const commands = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 ORDER BY seq ASC FOR UPDATE", [runId]);
       if (isTerminalRunStatus(run.status)) {
         await fenceActiveRunnerDispatchIntents(client, { runId }, reason);
-        return run;
+        return await this.releaseTerminalRunClaim(client, run);
       }
       const at = nowIso();
       const cancel = await this.createCancelRequest(client, { targetKind: "run", targetId: runId, run, command: null, reason, at, stage: "accepted" });
@@ -1153,9 +1159,10 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       }
       await fenceActiveRunnerDispatchIntents(client, { runId }, reason, at);
       const updated = await client.query(
-        `UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, updated_at = $3 WHERE id = $1 RETURNING *`,
+        `UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, claimed_by = NULL, lease_expires_at = NULL, updated_at = $3 WHERE id = $1 RETURNING *`,
         [runId, reason, at],
       );
+      await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [runId]);
       await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason, cancelRequestId: cancel.id, cancelEpoch: cancel.epoch });
       await this.appendCancelStage(client, runId, "terminalized", cancel);
       const next = runFromRow(updated.rows[0]);
@@ -1168,7 +1175,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     return this.withTransaction(async (client) => {
       const identity = await client.query("SELECT run_id FROM agentrun_commands WHERE id = $1", [commandId]);
       if (!identity.rows[0]) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
-      const run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id));
+      let run = await this.requireRunForUpdate(client, String(identity.rows[0].run_id));
+      if (isTerminalRunStatus(run.status)) run = await this.releaseTerminalRunClaim(client, run);
       const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
       const row = existing.rows[0];
       if (!row) throw new AgentRunError("schema-invalid", `command ${commandId} was not found`, { httpStatus: 404 });
@@ -1968,6 +1976,21 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     return session.activeRunId ?? session.lastRunId;
   }
 
+  private async releaseTerminalRunClaim(client: PoolClient, run: RunRecord): Promise<RunRecord> {
+    if (!isTerminalRunStatus(run.status)) return run;
+    return await this.releaseRunnerClaim(client, run);
+  }
+
+  private async releaseRunnerClaim(client: PoolClient, run: RunRecord): Promise<RunRecord> {
+    let released = run;
+    if (run.claimedBy !== null || run.leaseExpiresAt !== null) {
+      const result = await client.query("UPDATE agentrun_runs SET claimed_by = NULL, lease_expires_at = NULL, updated_at = $2 WHERE id = $1 RETURNING *", [run.id, nowIso()]);
+      released = runFromRow(result.rows[0]);
+    }
+    await client.query("DELETE FROM agentrun_leases WHERE run_id = $1", [run.id]);
+    return released;
+  }
+
   private async withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
     const client = await this.pool.connect();
     try {

src/mgr/runner-retention.ts

@@ -41,6 +41,7 @@ export interface RunnerRetentionSummary extends JsonRecord {
   deletedRunnerJobCount: number;
   deletedRunnerPodCount: number;
   deletedAssociatedResourceCount: number;
+  repairedTerminalClaimCount: number;
   activeRunRisk: boolean;
   reason: string;
   valuesPrinted: false;
@@ -71,6 +72,7 @@ export interface RunnerStalePendingReconcileSummary extends JsonRecord {
 }
 
 type CandidateKind = "idle" | "terminal" | "inactive" | "stale-pending";
+type RepairKind = "none" | "release-terminal-run-claim" | "release-terminal-command-claim" | "terminalize-stale-pending";
 
 interface RunnerPodObservation extends JsonRecord {
   name: string;
@@ -135,6 +137,7 @@ interface RunnerResourceEntry {
   protectedReason: string | null;
   candidateKind: CandidateKind;
   classificationReason: string;
+  repairKind: RepairKind;
   lastActiveAtMs: number;
   dbFactsFingerprint: string | null;
 }
@@ -188,10 +191,12 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
   let deletedRunnerJobCount = 0;
   let deletedRunnerPodCount = 0;
   let deletedAssociatedResourceCount = 0;
+  let repairedTerminalClaimCount = 0;
   for (const item of selected) {
     const retired = await retireSelectedRunner(input.store, input.options, item, `runner retention retired stale Kubernetes ${item.resourceKind}/${item.name}`);
     deletedRunnerJobCount += retired.deletedRunnerJobCount;
     deletedRunnerPodCount += retired.deletedRunnerPodCount;
+    repairedTerminalClaimCount += retired.repairedTerminalClaimCount;
     if (item.resourceKind === "job") deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
   }
   const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before;
@@ -210,6 +215,7 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
         deletedRunnerJobCount,
         deletedRunnerPodCount,
         deletedAssociatedResourceCount,
+        repairedTerminalClaimCount,
         selected: selected.map(resourceSummary),
         valuesPrinted: false,
       },
@@ -236,6 +242,7 @@ export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRu
     deletedRunnerJobCount,
     deletedRunnerPodCount,
     deletedAssociatedResourceCount,
+    repairedTerminalClaimCount,
     activeRunRisk: protectedActiveRunnerCount > 0,
     reason: requiredDeleteCount === 0 ? "within-limit" : "pre-create-retention",
     cleanupPolicy: cleanupPolicySummary(),
@@ -468,7 +475,7 @@ async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRe
   };
 }
 
-type ActivityFields = Pick<RunnerResourceEntry, "runId" | "commandId" | "runnerId" | "protectedActive" | "protectedReason" | "candidateKind" | "classificationReason" | "lastActiveAtMs" | "dbFactsFingerprint">;
+type ActivityFields = Pick<RunnerResourceEntry, "runId" | "commandId" | "runnerId" | "protectedActive" | "protectedReason" | "candidateKind" | "classificationReason" | "repairKind" | "lastActiveAtMs" | "dbFactsFingerprint">;
 
 async function activityFor(store: AgentRunStore, input: {
   runId: string | null;
@@ -506,8 +513,21 @@ async function activityFor(store: AgentRunStore, input: {
   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 (isTerminalRunStatus(run.status)) {
+    const claimDrift = run.claimedBy !== null || run.leaseExpiresAt !== null;
+    return candidateActivity(input, "terminal", claimDrift ? "terminal-run-stale-claim" : "terminal-run", lastActiveAtMs, dbFactsFingerprint, claimDrift ? "release-terminal-run-claim" : "none");
+  }
+  if (isTerminalCommandState(command.state)) {
+    const claimDrift = input.terminal && (run.claimedBy !== null || run.leaseExpiresAt !== null);
+    return candidateActivity(
+      input,
+      input.terminal ? "terminal" : "idle",
+      claimDrift ? "terminal-command-stale-claim" : input.terminal ? "terminal-runner-resource" : "terminal-command-idle-resource",
+      lastActiveAtMs,
+      dbFactsFingerprint,
+      claimDrift ? "release-terminal-command-claim" : "none",
+    );
+  }
   if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs, dbFactsFingerprint);
 
   const podBlockReason = stalePendingPodBlockReason(input.podObservations);
@@ -515,12 +535,13 @@ async function activityFor(store: AgentRunStore, input: {
   if (!hasNoBusinessProgress(run, command, commands, events, input.stalePendingMaxAgeMs)) {
     return protectedActivity(input, "business-progress-observed-or-unverifiable", lastActiveAtMs, dbFactsFingerprint);
   }
-  return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint);
+  return candidateActivity(input, "stale-pending", "stale-pending-no-business-progress", lastActiveAtMs, dbFactsFingerprint, "terminalize-stale-pending");
 }
 
 interface RetiredRunnerResource extends JsonRecord {
   deletedRunnerJobCount: number;
   deletedRunnerPodCount: number;
+  repairedTerminalClaimCount: number;
 }
 
 async function retireSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, item: RunnerResourceEntry, reason: string): Promise<RetiredRunnerResource> {
@@ -535,6 +556,7 @@ async function retireSelectedRunner(store: AgentRunStore, options: RunnerRetenti
     runId: item.runId,
     commandId: item.commandId,
     terminalizeStalePending: item.candidateKind === "stale-pending",
+    releaseRunnerClaim: item.repairKind === "release-terminal-run-claim" || item.repairKind === "release-terminal-command-claim",
     reason,
   }, async (facts) => {
     const currentDbFingerprint = databaseFactsFingerprint(facts.run, facts.command, facts.commands, facts.events);
@@ -542,9 +564,13 @@ async function retireSelectedRunner(store: AgentRunStore, options: RunnerRetenti
     await deleteRunnerResourceCas(options, item);
   });
   if (item.resourceKind === "job") {
-    return { deletedRunnerJobCount: 1, deletedRunnerPodCount: 0 };
+    return { deletedRunnerJobCount: 1, deletedRunnerPodCount: 0, repairedTerminalClaimCount: isClaimRepair(item.repairKind) ? 1 : 0 };
   }
-  return { deletedRunnerJobCount: 0, deletedRunnerPodCount: 1 };
+  return { deletedRunnerJobCount: 0, deletedRunnerPodCount: 1, repairedTerminalClaimCount: isClaimRepair(item.repairKind) ? 1 : 0 };
+}
+
+function isClaimRepair(repairKind: RepairKind): boolean {
+  return repairKind === "release-terminal-run-claim" || repairKind === "release-terminal-command-claim";
 }
 
 async function revalidateSelectedRunner(store: AgentRunStore, options: RunnerRetentionOptions, selected: RunnerResourceEntry): Promise<void> {
@@ -725,6 +751,7 @@ function resourceSummary(item: RunnerResourceEntry): JsonRecord {
     protectedReason: item.protectedReason,
     candidateKind: item.candidateKind,
     classificationReason: item.classificationReason,
+    repairKind: item.repairKind,
     terminal: item.terminal,
     podPhase: item.podPhase,
     podObservations: item.podObservations.slice(0, 5).map(podObservationSummary),
@@ -747,7 +774,8 @@ function cleanupPolicySummary(): JsonRecord {
   return {
     mode: "db-ledger-and-k8s-observation",
     forceActive: false,
-    activeProtection: "protect non-terminal DB run/command unless unclaimed no-progress facts and stale Pending Pod observations are complete",
+    activeProtection: "protect fresh claims and active Pods; terminal Pods do not remain active merely because their container once started",
+    terminalClaimRepair: "release stale terminal run claims under the same DB and Kubernetes CAS fence",
     deletionOrder: "idle, terminal, stale-pending, inactive by lastActiveAt",
     valuesPrinted: false,
   };
@@ -874,7 +902,14 @@ function stalePendingPodBlockReason(observations: RunnerPodObservation[]): strin
 }
 
 function activePodProtectionReason(observations: RunnerPodObservation[]): string | null {
-  if (observations.some((item) => item.phase === "Running" || item.ready || item.runnerContainerReady || item.runnerContainerStarted || item.runnerContainerState === "running" || item.runnerContainerState === "terminated")) return "runner-pod-started-running-or-ready";
+  if (observations.some((item) => {
+    const terminalPhase = item.phase === "Succeeded" || item.phase === "Failed";
+    return item.phase === "Running"
+      || item.ready
+      || item.runnerContainerReady
+      || item.runnerContainerState === "running"
+      || (!terminalPhase && (item.runnerContainerStarted || item.runnerContainerState === "terminated"));
+  })) return "runner-pod-started-running-or-ready";
   return null;
 }
 
@@ -907,11 +942,11 @@ function hasNoBusinessProgress(run: RunRecord, command: CommandRecord, commands:
 }
 
 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 };
+  return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: reason, candidateKind: "inactive", classificationReason: reason, repairKind: "none", lastActiveAtMs, dbFactsFingerprint };
 }
 
-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 candidateActivity(input: { runId: string | null; commandId: string | null; runnerId: string | null }, kind: CandidateKind, reason: string, lastActiveAtMs: number, dbFactsFingerprint: string, repairKind: RepairKind = "none"): ActivityFields {
+  return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind: kind, classificationReason: reason, repairKind, lastActiveAtMs, dbFactsFingerprint };
 }
 
 function databaseFactsFingerprint(run: RunRecord, command: CommandRecord, commands: CommandRecord[], events: import("../common/types.js").RunEvent[]): string {
@@ -982,6 +1017,7 @@ function runnerSafetyFingerprint(item: RunnerResourceEntry): string {
     protectedReason: item.protectedReason,
     candidateKind: item.candidateKind,
     classificationReason: item.classificationReason,
+    repairKind: item.repairKind,
     lastActiveAtMs: item.lastActiveAtMs,
     dbFactsFingerprint: item.dbFactsFingerprint,
     podObservations: [...item.podObservations].sort((left, right) => left.name.localeCompare(right.name)).map(podSafetyFacts),
@@ -1039,6 +1075,7 @@ function resourceIdentitySummary(item: RunnerResourceEntry): JsonRecord {
     runnerId: item.runnerId,
     candidateKind: item.candidateKind,
     classificationReason: item.classificationReason,
+    repairKind: item.repairKind,
     valuesPrinted: false,
   };
 }

src/mgr/store.ts

@@ -43,6 +43,7 @@ export interface RunnerRetentionFenceInput {
   runId: string;
   commandId: string;
   terminalizeStalePending: boolean;
+  releaseRunnerClaim: boolean;
   reason: string;
 }
 
@@ -515,6 +516,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
       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);
+      else if (input.releaseRunnerClaim && (isTerminalRunStatus(run.status) || isTerminalCommandState(command.state))) this.releaseRunnerClaim(run);
       return result;
     } finally {
       this.runnerRetentionFences.delete(input.runId);
@@ -550,15 +552,16 @@ export class MemoryAgentRunStore implements AgentRunStore {
 
   finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): CommandRecord {
     const command = this.getCommand(commandId);
+    let run = this.getRun(command.runId);
+    if (isTerminalRunStatus(run.status)) run = this.releaseTerminalRunClaim(run);
     if (isTerminalCommandState(command.state)) {
       this.cancelRunnerDispatchIntentForCommand(commandId, "command is terminal");
-      if (command.state === "cancelled" && result.terminalStatus !== "cancelled") this.appendLateWriteRejected(command.runId, command, result, "command-status");
+      if (command.state === "cancelled" && result.terminalStatus !== "cancelled") this.appendLateWriteRejected(run.id, command, result, "command-status");
       return command;
     }
     const next = { ...command, state: commandStateFromTerminal(result.terminalStatus), updatedAt: nowIso() };
     this.commands.set(commandId, next);
     this.cancelRunnerDispatchIntentForCommand(commandId, `command terminalized as ${next.state}`, next.updatedAt);
-    const run = this.getRun(command.runId);
     if (result.threadId && run.sessionRef?.sessionId) this.upsertSessionThread(run, result.threadId, result.turnId ?? null);
     if (command.type === "turn") this.touchSessionForRun(this.getRun(command.runId), { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: next.updatedAt }, { bumpVersion: true, at: next.updatedAt });
     this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: next.state, terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null });
@@ -605,11 +608,12 @@ export class MemoryAgentRunStore implements AgentRunStore {
     const existing = this.getRun(runId);
     if (isTerminalRunStatus(existing.status)) {
       this.cancelRunnerDispatchIntentsForRun(runId, "run is terminal");
-      if (existing.status === "cancelled" && result.terminalStatus !== "cancelled") this.appendLateWriteRejected(runId, null, result, "run-status");
-      return existing;
+      const released = this.releaseTerminalRunClaim(existing);
+      if (released.status === "cancelled" && result.terminalStatus !== "cancelled") this.appendLateWriteRejected(runId, null, result, "run-status");
+      return released;
     }
     const status = statusFromTerminal(result.terminalStatus);
-    const next = this.updateRun(runId, { status, terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: result.failureMessage });
+    const next = this.updateRun(runId, { status, terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: result.failureMessage, claimedBy: null, leaseExpiresAt: null });
     this.cancelRunnerDispatchIntentsForRun(runId, `run terminalized as ${status}`, next.updatedAt);
     if (result.threadId && next.sessionRef?.sessionId) this.upsertSessionThread(next, result.threadId, result.turnId ?? null);
     this.appendEvent(runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage });
@@ -621,7 +625,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
     const run = this.getRun(runId);
     if (isTerminalRunStatus(run.status)) {
       this.cancelRunnerDispatchIntentsForRun(runId, reason);
-      return run;
+      return this.releaseTerminalRunClaim(run);
     }
     const at = nowIso();
     const cancel = this.createCancelRequest({ targetKind: "run", targetId: runId, run, command: null, reason, at, stage: "accepted" });
@@ -640,7 +644,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
       this.appendEvent(runId, "backend_status", { phase: "command-terminal", commandId: command.id, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", cancelRequestId: cancel.id, cancelEpoch: cancel.epoch });
     }
     this.cancelRunnerDispatchIntentsForRun(runId, reason, at);
-    const next = this.updateRun(runId, { status: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: reason });
+    const next = this.updateRun(runId, { status: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: reason, claimedBy: null, leaseExpiresAt: null });
     this.appendEvent(runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason, cancelRequestId: cancel.id, cancelEpoch: cancel.epoch });
     this.appendCancelStage(runId, "terminalized", cancel);
     this.touchSessionForRun(next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: "cancelled", failureKind: "cancelled", lastActivityAt: next.updatedAt }, { bumpVersion: true, at: next.updatedAt });
@@ -649,11 +653,12 @@ export class MemoryAgentRunStore implements AgentRunStore {
 
   cancelCommand(commandId: string, reason = "cancel requested"): CommandRecord {
     const command = this.getCommand(commandId);
+    const run = this.getRun(command.runId);
+    if (isTerminalRunStatus(run.status)) this.releaseTerminalRunClaim(run);
     if (isTerminalCommandState(command.state)) {
       this.cancelRunnerDispatchIntentForCommand(commandId, reason);
       return command;
     }
-    const run = this.getRun(command.runId);
     const at = nowIso();
     const cancel = this.createCancelRequest({ targetKind: "command", targetId: commandId, run, command, reason, at, stage: "accepted" });
     this.appendCancelStage(command.runId, "accepted", cancel);
@@ -1046,6 +1051,16 @@ export class MemoryAgentRunStore implements AgentRunStore {
     return next;
   }
 
+  private releaseTerminalRunClaim(run: RunRecord): RunRecord {
+    if (!isTerminalRunStatus(run.status)) return run;
+    return this.releaseRunnerClaim(run);
+  }
+
+  private releaseRunnerClaim(run: RunRecord): RunRecord {
+    if (run.claimedBy === null && run.leaseExpiresAt === null) return run;
+    return this.updateRun(run.id, { claimedBy: null, leaseExpiresAt: null });
+  }
+
   private terminalizeRunnerRetention(run: RunRecord, commands: CommandRecord[], reason: string): void {
     const at = nowIso();
     for (const command of commands.filter((item) => !isTerminalCommandState(item.state))) {

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

@@ -93,11 +93,13 @@ class RetentionFactStore extends MemoryAgentRunStore {
     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) {
+    if (input.terminalizeStalePending && !["completed", "failed", "blocked", "cancelled"].includes(fact.run.status)) {
       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 };
+    } else if (input.releaseRunnerClaim && (["completed", "failed", "blocked", "cancelled"].includes(fact.run.status) || ["completed", "failed", "cancelled"].includes(fact.command.state))) {
+      fact.run = { ...fact.run, claimedBy: null, leaseExpiresAt: null, updatedAt: new Date().toISOString() };
     }
     return result;
   }
@@ -117,6 +119,9 @@ const selfTest: SelfTestCase = async (context) => {
     await assertTwentyStalePendingSelectsOnlyOldest({ fixturePath, statePath, kubectlCommand });
     await assertLegacyNoUserMessageCandidate({ fixturePath, statePath, kubectlCommand });
     await assertEventOnlyFailedMountCandidate({ fixturePath, statePath, kubectlCommand });
+    await assertTerminalClaimDriftCandidate({ fixturePath, statePath, kubectlCommand });
+    await assertTerminalCommandClaimDriftCandidate({ fixturePath, statePath, kubectlCommand });
+    await assertTerminalClaimRepairCasConflictFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertProtectedMatrix({ fixturePath, statePath, kubectlCommand });
     await assertPendingToRunningRaceFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertDbClaimAfterRecheckFailsClosed({ fixturePath, statePath, kubectlCommand });
@@ -157,6 +162,10 @@ const selfTest: SelfTestCase = async (context) => {
       "pod-identity-mismatch-protected",
       "event-facts-unavailable-protected",
       "terminal-ledger-running-pod-protected",
+      "terminal-runner-stale-claim-typed-repair",
+      "terminal-command-runner-stale-claim-typed-repair",
+      "terminal-claim-repair-cas-conflict-fails-closed",
+      "terminal-runner-pod-completed-is-not-active",
       "selected-runner-race-fails-closed",
       "db-claim-after-recheck-fails-closed",
       "pod-resource-version-cas-fails-closed",
@@ -254,6 +263,18 @@ async function assertProtectedMatrix(input: { fixturePath: string; statePath: st
   terminalRunning.run.terminalStatus = "completed";
   terminalRunning.command.state = "completed";
   await assertProtected(input, [terminalRunning], [runnerPod(terminalRunning, 105, "running")], "runner-pod-started-running-or-ready");
+  const terminalFreshClaim = pendingFact(107, oldAt, "claimed");
+  terminalFreshClaim.run.status = "completed";
+  terminalFreshClaim.run.terminalStatus = "completed";
+  terminalFreshClaim.command.state = "completed";
+  const terminalFreshClaimPod = runnerPod(terminalFreshClaim, 107, "running");
+  terminalFreshClaimPod.status = {
+    phase: "Succeeded",
+    startTime: terminalFreshClaim.run.createdAt,
+    conditions: [{ type: "Ready", status: "False", reason: "PodCompleted" }],
+    containerStatuses: [{ name: "runner", ready: false, started: false, restartCount: 0, state: { terminated: { exitCode: 0, reason: "Completed" } } }],
+  };
+  await assertProtected(input, [terminalFreshClaim], [terminalFreshClaimPod], "fresh-active-run");
   const identityMismatchPod = runnerPod(base, 100, "pending");
   ((identityMismatchPod.metadata as JsonRecord).annotations as JsonRecord)["agentrun.pikastech.local/command-id"] = "cmd-mismatched";
   await assertProtected(input, [base], [identityMismatchPod], "pod-identity-mismatch");
@@ -297,6 +318,97 @@ async function assertEventOnlyFailedMountCandidate(input: { fixturePath: string;
   assert.equal((await readDeletionState(input.statePath)).length, 0);
 }
 
+async function assertTerminalClaimDriftCandidate(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(88, new Date(Date.now() - 60 * 60_000).toISOString(), "claimed");
+  fact.run.status = "failed";
+  fact.run.terminalStatus = "failed";
+  fact.run.failureKind = "infra-failed";
+  fact.run.failureMessage = "terminal claim drift self-test";
+  fact.run.leaseExpiresAt = new Date(Date.now() - 30 * 60_000).toISOString();
+  fact.command.state = "failed";
+  const job = runnerJob(fact, 88);
+  job.status = { conditions: [{ type: "Failed", status: "True", reason: "BackoffLimitExceeded" }] };
+  const pod = runnerPod(fact, 88, "running");
+  pod.status = {
+    phase: "Succeeded",
+    startTime: fact.run.createdAt,
+    conditions: [{ type: "Ready", status: "False", reason: "PodCompleted" }],
+    containerStatuses: [{
+      name: "runner",
+      ready: false,
+      started: false,
+      restartCount: 0,
+      state: { terminated: { exitCode: 0, reason: "Completed", finishedAt: new Date(Date.now() - 45 * 60_000).toISOString() } },
+    }],
+  };
+  await writeFixture(input, { jobs: [job], pods: [pod], events: [] });
+  const store = new RetentionFactStore([fact]);
+  const summary = await enforceRunnerRetentionBeforeCreate({ store, options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 });
+  assert.equal(summary.inactiveCandidateCount, 1);
+  assert.equal(summary.selectedRunnerCount, 1);
+  const selected = firstSummary(summary, "selected");
+  assert.equal(selected.candidateKind, "terminal");
+  assert.equal(selected.classificationReason, "terminal-run-stale-claim");
+  assert.equal(selected.repairKind, "release-terminal-run-claim");
+  assert.equal(summary.repairedTerminalClaimCount, 1);
+  assert.equal(store.getRun(fact.run.id).claimedBy, null);
+  assert.equal(store.getRun(fact.run.id).leaseExpiresAt, null);
+  assert.deepEqual(deletionOperations(await readDeletionState(input.statePath)).filter((item) => item.resource === "job").map((item) => item.name), ["agentrun-v02-runner-stale-88"]);
+}
+
+async function assertTerminalCommandClaimDriftCandidate(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(87, new Date(Date.now() - 60 * 60_000).toISOString(), "claimed");
+  fact.run.leaseExpiresAt = new Date(Date.now() - 30 * 60_000).toISOString();
+  fact.command.state = "failed";
+  const job = runnerJob(fact, 87);
+  job.status = { conditions: [{ type: "Failed", status: "True", reason: "BackoffLimitExceeded" }] };
+  const pod = runnerPod(fact, 87, "running");
+  pod.status = {
+    phase: "Failed",
+    startTime: fact.run.createdAt,
+    conditions: [{ type: "Ready", status: "False", reason: "PodFailed" }],
+    containerStatuses: [{ name: "runner", ready: false, started: false, restartCount: 0, state: { terminated: { exitCode: 1, reason: "Error" } } }],
+  };
+  await writeFixture(input, { jobs: [job], pods: [pod], events: [] });
+  const store = new RetentionFactStore([fact]);
+  const summary = await enforceRunnerRetentionBeforeCreate({ store, options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 });
+  const selected = firstSummary(summary, "selected");
+  assert.equal(selected.candidateKind, "terminal");
+  assert.equal(selected.classificationReason, "terminal-command-stale-claim");
+  assert.equal(selected.repairKind, "release-terminal-command-claim");
+  assert.equal(summary.repairedTerminalClaimCount, 1);
+  assert.equal(store.getRun(fact.run.id).status, "claimed");
+  assert.equal(store.getRun(fact.run.id).claimedBy, null);
+  assert.equal(store.getRun(fact.run.id).leaseExpiresAt, null);
+}
+
+async function assertTerminalClaimRepairCasConflictFailsClosed(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const fact = pendingFact(86, new Date(Date.now() - 60 * 60_000).toISOString(), "claimed");
+  fact.run.status = "failed";
+  fact.run.terminalStatus = "failed";
+  fact.run.failureKind = "infra-failed";
+  fact.run.leaseExpiresAt = new Date(Date.now() - 30 * 60_000).toISOString();
+  fact.command.state = "failed";
+  const job = runnerJob(fact, 86);
+  job.status = { conditions: [{ type: "Failed", status: "True", reason: "BackoffLimitExceeded" }] };
+  const pod = runnerPod(fact, 86, "running");
+  pod.status = {
+    phase: "Succeeded",
+    startTime: fact.run.createdAt,
+    conditions: [{ type: "Ready", status: "False", reason: "PodCompleted" }],
+    containerStatuses: [{ name: "runner", ready: false, started: false, restartCount: 0, state: { terminated: { exitCode: 0, reason: "Completed" } } }],
+  };
+  const racedPod = structuredClone(pod);
+  (racedPod.metadata as JsonRecord).resourceVersion = "2";
+  await writeFixture(input, { jobs: [job], pods: [pod], events: [], raceOnCasDelete: true, racePods: [racedPod] });
+  const store = new RetentionFactStore([fact]);
+  const error = await retentionError(() => enforceRunnerRetentionBeforeCreate({ store, options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 1 }));
+  assert.equal(error.details?.reason, "runner-retention-k8s-cas-rejected");
+  assert.equal(store.getRun(fact.run.id).claimedBy, `runner-retention-${String(86).padStart(2, "0")}`);
+  assert.notEqual(store.getRun(fact.run.id).leaseExpiresAt, null);
+  assert.equal(deletionOperations(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");
@@ -411,7 +523,7 @@ async function assertMemoryDbFenceRejectsConcurrentClaim(): Promise<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 () => {
+  const fenced = Promise.resolve(store.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, releaseRunnerClaim: false, reason: "retention memory self-test" }, async () => {
     enterFence();
     await release;
   }));

src/selftest/cases/23-runner-terminal-lifecycle.ts

@@ -0,0 +1,73 @@
+import assert from "node:assert/strict";
+import { MemoryAgentRunStore } from "../../mgr/store.js";
+import type { FailureKind, TerminalStatus } from "../../common/types.js";
+import type { SelfTestCase } from "../harness.js";
+
+const selfTest: SelfTestCase = () => {
+  const store = new MemoryAgentRunStore();
+  for (const scenario of [
+    { name: "completed", terminalStatus: "completed" as const, failureKind: null },
+    { name: "failed", terminalStatus: "failed" as const, failureKind: "infra-failed" as const },
+    { name: "timeout", terminalStatus: "failed" as const, failureKind: "backend-timeout" as const },
+  ]) assertFinishedRunReleasesClaim(store, scenario);
+  assertCancelledRunReleasesClaim(store);
+  return {
+    name: "runner-terminal-lifecycle",
+    tests: [
+      "terminal-run-releases-runner-claim",
+      "terminal-run-claim-release-idempotent",
+      "terminal-command-keeps-reusable-run-claim",
+      "cancelled-run-releases-runner-claim",
+      "timeout-run-releases-runner-claim",
+    ],
+  };
+};
+
+function assertFinishedRunReleasesClaim(store: MemoryAgentRunStore, scenario: { name: string; terminalStatus: TerminalStatus; failureKind: FailureKind | null }): void {
+  const run = createLifecycleRun(store, scenario.name);
+  const command = store.createCommand(run.id, { type: "turn", payload: { prompt: scenario.name }, idempotencyKey: `terminal-lifecycle-${scenario.name}` });
+  const runner = store.registerRunner({ id: `runner-terminal-${scenario.name}`, runId: run.id, backendProfile: "codex" });
+  store.claimRun(run.id, runner.id, 60_000);
+  store.finishCommand(command.id, { terminalStatus: scenario.terminalStatus, failureKind: scenario.failureKind, failureMessage: scenario.failureKind });
+  assert.equal(store.getRun(run.id).claimedBy, runner.id, "a terminal command must not release a reusable non-terminal run claim");
+  const terminal = store.finishRun(run.id, { terminalStatus: scenario.terminalStatus, failureKind: scenario.failureKind, failureMessage: scenario.failureKind });
+  assert.equal(terminal.claimedBy, null);
+  assert.equal(terminal.leaseExpiresAt, null);
+
+  terminal.claimedBy = `runner-drift-${scenario.name}`;
+  terminal.leaseExpiresAt = new Date(Date.now() - 60_000).toISOString();
+  const replay = store.finishRun(run.id, { terminalStatus: scenario.terminalStatus, failureKind: scenario.failureKind, failureMessage: scenario.failureKind });
+  assert.equal(replay.claimedBy, null);
+  assert.equal(replay.leaseExpiresAt, null);
+  assert.equal(store.listEvents(run.id, 0, 500).filter((event) => event.type === "terminal_status").length, 1);
+}
+
+function assertCancelledRunReleasesClaim(store: MemoryAgentRunStore): void {
+  const run = createLifecycleRun(store, "cancelled");
+  const command = store.createCommand(run.id, { type: "turn", payload: { prompt: "cancelled" }, idempotencyKey: "terminal-lifecycle-cancelled" });
+  const runner = store.registerRunner({ id: "runner-terminal-cancelled", runId: run.id, backendProfile: "codex" });
+  store.claimRun(run.id, runner.id, 60_000);
+  const cancelled = store.cancelRun(run.id, "terminal lifecycle self-test");
+  assert.equal(store.getCommand(command.id).state, "cancelled");
+  assert.equal(cancelled.claimedBy, null);
+  assert.equal(cancelled.leaseExpiresAt, null);
+  cancelled.claimedBy = "runner-cancelled-drift";
+  cancelled.leaseExpiresAt = new Date(Date.now() - 60_000).toISOString();
+  const replay = store.cancelRun(run.id, "terminal lifecycle duplicate self-test");
+  assert.equal(replay.claimedBy, null);
+  assert.equal(replay.leaseExpiresAt, null);
+}
+
+function createLifecycleRun(store: MemoryAgentRunStore, suffix: string) {
+  return store.createRun({
+    tenantId: "selftest",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: `/tmp/agentrun-terminal-${suffix}` },
+    providerId: "NC01",
+    backendProfile: "codex",
+    executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } },
+    traceSink: null,
+  });
+}
+
+export default selfTest;

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

@@ -1,4 +1,5 @@
 import assert from "node:assert/strict";
+import { Pool } from "pg";
 import { AgentRunError } from "../../common/errors.js";
 import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js";
 
@@ -7,6 +8,7 @@ if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is requir
 
 const primary = await createPostgresAgentRunStore({ connectionString });
 const concurrent = await createPostgresAgentRunStore({ connectionString });
+const inspector = new Pool({ connectionString });
 
 try {
   await assertNamespaceCreateFence();
@@ -14,9 +16,12 @@ try {
   await assertRetentionFenceBlocksClaim();
   await assertRetentionTerminalizationIsIdempotent();
   await assertRetentionTerminalizationRollsBackOnCasFailure();
-  console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-independent-create-fences-do-not-serialize", "postgres-run-retention-fence-blocks-claim", "postgres-run-retention-terminalization-idempotent", "postgres-run-retention-cas-failure-rolls-back-terminalization"] }));
+  await assertTerminalLifecycleReleasesClaim();
+  await assertTerminalHeartbeatRaceReleasesClaim();
+  await assertHistoricalTerminalClaimRepairIsFenced();
+  console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-independent-create-fences-do-not-serialize", "postgres-run-retention-fence-blocks-claim", "postgres-run-retention-terminalization-idempotent", "postgres-run-retention-cas-failure-rolls-back-terminalization", "postgres-terminal-run-atomically-releases-claim-and-lease", "postgres-terminal-replay-repairs-claim-after-restart", "postgres-terminal-heartbeat-race-releases-claim", "postgres-retention-terminal-claim-repair-fenced"] }));
 } finally {
-  await Promise.all([primary.close(), concurrent.close()]);
+  await Promise.all([primary.close(), concurrent.close(), inspector.end()]);
 }
 
 async function assertNamespaceCreateFence(): Promise<void> {
@@ -80,7 +85,7 @@ async function assertRetentionFenceBlocksClaim(): Promise<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) => {
+  const fenced = primary.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, releaseRunnerClaim: false, reason: "postgres retention fence self-test" }, async (facts) => {
     assert.equal(facts.run.id, run.id);
     assert.equal(facts.command.id, command.id);
     fenceEntered();
@@ -113,7 +118,7 @@ async function assertRetentionTerminalizationIsIdempotent(): Promise<void> {
     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" };
+  const fenceInput = { runId: run.id, commandId: command.id, terminalizeStalePending: true, releaseRunnerClaim: false, reason: "postgres retention idempotence self-test" };
   await primary.withRunnerRetentionFence(fenceInput, async (facts) => {
     assert.equal(facts.run.status, "pending");
   });
@@ -140,7 +145,7 @@ async function assertRetentionTerminalizationRollsBackOnCasFailure(): Promise<vo
   });
   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 () => {
+    primary.withRunnerRetentionFence({ runId: run.id, commandId: command.id, terminalizeStalePending: true, releaseRunnerClaim: false, 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",
@@ -151,6 +156,135 @@ async function assertRetentionTerminalizationRollsBackOnCasFailure(): Promise<vo
   assert.equal(events.filter((event) => event.payload.retentionFence === true).length, 0);
 }
 
+async function assertTerminalLifecycleReleasesClaim(): Promise<void> {
+  for (const scenario of [
+    { name: "completed", terminalStatus: "completed" as const, failureKind: null },
+    { name: "failed", terminalStatus: "failed" as const, failureKind: "infra-failed" as const },
+    { name: "timeout", terminalStatus: "failed" as const, failureKind: "backend-timeout" as const },
+  ]) {
+    const { runId, commandId, runnerId } = await createClaimedRun(`postgres-terminal-${scenario.name}`);
+    await primary.finishCommand(commandId, { terminalStatus: scenario.terminalStatus, failureKind: scenario.failureKind, failureMessage: scenario.failureKind });
+    assert.equal((await primary.getRun(runId)).claimedBy, runnerId, "a terminal command must keep the reusable non-terminal run claim");
+    await primary.finishRun(runId, { terminalStatus: scenario.terminalStatus, failureKind: scenario.failureKind, failureMessage: scenario.failureKind });
+    await assertClaimReleased(runId);
+
+    await injectTerminalClaimDrift(runId, `runner-restart-drift-${scenario.name}`);
+    const restarted = await createPostgresAgentRunStore({ connectionString });
+    try {
+      await restarted.finishRun(runId, { terminalStatus: scenario.terminalStatus, failureKind: scenario.failureKind, failureMessage: scenario.failureKind });
+    } finally {
+      await restarted.close();
+    }
+    await assertClaimReleased(runId);
+    const events = await primary.listEvents(runId, 0, 500);
+    assert.equal(events.filter((event) => event.type === "terminal_status").length, 1);
+  }
+
+  const cancelled = await createClaimedRun("postgres-terminal-cancelled");
+  await primary.cancelRun(cancelled.runId, "postgres terminal lifecycle self-test");
+  await assertClaimReleased(cancelled.runId);
+  await injectTerminalClaimDrift(cancelled.runId, "runner-restart-drift-cancelled");
+  await concurrent.cancelRun(cancelled.runId, "postgres terminal lifecycle duplicate self-test");
+  await assertClaimReleased(cancelled.runId);
+}
+
+async function assertTerminalHeartbeatRaceReleasesClaim(): Promise<void> {
+  const claimed = await createClaimedRun("postgres-terminal-heartbeat-race");
+  const terminal = primary.finishRun(claimed.runId, { terminalStatus: "failed", failureKind: "backend-timeout", failureMessage: "terminal heartbeat race self-test" });
+  const heartbeat = concurrent.heartbeat(claimed.runId, claimed.runnerId, 60_000);
+  const results = await Promise.allSettled([terminal, heartbeat]);
+  assert.equal(results[0]?.status, "fulfilled");
+  assert.equal(results[1]?.status, "fulfilled");
+  await assertClaimReleased(claimed.runId);
+}
+
+async function assertHistoricalTerminalClaimRepairIsFenced(): Promise<void> {
+  const claimed = await createClaimedRun("postgres-terminal-retention-repair");
+  await primary.finishCommand(claimed.commandId, { terminalStatus: "failed", failureKind: "infra-failed", failureMessage: "retention repair self-test" });
+  await primary.finishRun(claimed.runId, { terminalStatus: "failed", failureKind: "infra-failed", failureMessage: "retention repair self-test" });
+  await injectTerminalClaimDrift(claimed.runId, "runner-retention-drift");
+  const terminalRepairInput = {
+    runId: claimed.runId,
+    commandId: claimed.commandId,
+    terminalizeStalePending: false,
+    releaseRunnerClaim: true,
+    reason: "postgres terminal claim repair self-test",
+  };
+  await assert.rejects(
+    primary.withRunnerRetentionFence(terminalRepairInput, async () => {
+      throw new AgentRunError("infra-failed", "simulated terminal claim 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",
+  );
+  await assertClaimPresent(claimed.runId, "runner-retention-drift");
+  await primary.withRunnerRetentionFence(terminalRepairInput, async (facts) => {
+    assert.equal(facts.run.claimedBy, "runner-retention-drift");
+    assert.equal(facts.run.status, "failed");
+  });
+  await assertClaimReleased(claimed.runId);
+
+  const commandTerminal = await createClaimedRun("postgres-terminal-command-retention-repair");
+  await primary.finishCommand(commandTerminal.commandId, { terminalStatus: "failed", failureKind: "infra-failed", failureMessage: "terminal command retention repair self-test" });
+  await injectTerminalClaimDrift(commandTerminal.runId, "runner-terminal-command-drift");
+  await primary.withRunnerRetentionFence({
+    runId: commandTerminal.runId,
+    commandId: commandTerminal.commandId,
+    terminalizeStalePending: false,
+    releaseRunnerClaim: true,
+    reason: "postgres terminal command claim repair self-test",
+  }, async (facts) => {
+    assert.equal(facts.run.status, "claimed");
+    assert.equal(facts.command.state, "failed");
+    assert.equal(facts.run.claimedBy, "runner-terminal-command-drift");
+  });
+  assert.equal((await primary.getRun(commandTerminal.runId)).status, "claimed");
+  await assertClaimReleased(commandTerminal.runId);
+}
+
+async function createClaimedRun(suffix: string): Promise<{ runId: string; commandId: string; runnerId: string }> {
+  const run = await primary.createRun({
+    tenantId: "selftest",
+    projectId: "pikasTech/agentrun",
+    workspaceRef: { kind: "host-path", path: `/tmp/${suffix}` },
+    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: suffix } });
+  const runnerId = `runner-${run.id}`;
+  await primary.registerRunner({ id: runnerId, runId: run.id, backendProfile: "codex" });
+  await primary.claimRun(run.id, runnerId, 60_000);
+  return { runId: run.id, commandId: command.id, runnerId };
+}
+
+async function injectTerminalClaimDrift(runId: string, runnerId: string): Promise<void> {
+  const leaseExpiresAt = new Date(Date.now() - 60_000).toISOString();
+  await inspector.query("UPDATE agentrun_runs SET claimed_by = $2, lease_expires_at = $3 WHERE id = $1", [runId, runnerId, leaseExpiresAt]);
+  await inspector.query(
+    `INSERT INTO agentrun_leases (run_id, runner_id, lease_expires_at, stale_recovery_marker, updated_at)
+     VALUES ($1, $2, $3, NULL, now())
+     ON CONFLICT (run_id) DO UPDATE SET runner_id = EXCLUDED.runner_id, lease_expires_at = EXCLUDED.lease_expires_at, updated_at = now()`,
+    [runId, runnerId, leaseExpiresAt],
+  );
+}
+
+async function assertClaimReleased(runId: string): Promise<void> {
+  const run = await primary.getRun(runId);
+  assert.equal(run.claimedBy, null);
+  assert.equal(run.leaseExpiresAt, null);
+  const leases = await inspector.query("SELECT count(*)::int AS count FROM agentrun_leases WHERE run_id = $1", [runId]);
+  assert.equal(Number(leases.rows[0]?.count ?? -1), 0);
+}
+
+async function assertClaimPresent(runId: string, runnerId: string): Promise<void> {
+  const run = await primary.getRun(runId);
+  assert.equal(run.claimedBy, runnerId);
+  assert.notEqual(run.leaseExpiresAt, null);
+  const leases = await inspector.query("SELECT count(*)::int AS count FROM agentrun_leases WHERE run_id = $1", [runId]);
+  assert.equal(Number(leases.rows[0]?.count ?? -1), 1);
+}
+
 async function delay(ms: number): Promise<void> {
   await new Promise<void>((resolve) => setTimeout(resolve, ms));
 }