Merge pull request #333 from pikasTech/fix/332-terminal-runner-retention

修复终态 runner 被历史活性事实占用容量

src/mgr/runner-retention.ts

@@ -256,11 +256,10 @@ export async function reconcileStalePendingRunners(input: { store: AgentRunStore
     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 inactiveCandidates = resources.filter((item) => !item.protectedActive).sort(compareCandidates);
+    const stalePendingCandidates = inactiveCandidates.filter((item) => item.candidateKind === "stale-pending");
     const protectedActiveRunners = resources.filter((item) => item.protectedActive);
-    const selected = stalePendingCandidates.slice(0, limit);
+    const selected = inactiveCandidates.slice(0, limit);
     const outcomes: JsonRecord[] = [];
     let terminalizedRunCount = 0;
     let terminalizedCommandCount = 0;
@@ -297,6 +296,7 @@ export async function reconcileStalePendingRunners(input: { store: AgentRunStore
       }
     }
     const candidateSummaries = boundedResourceSummaries(stalePendingCandidates);
+    const inactiveCandidateSummaries = boundedResourceSummaries(inactiveCandidates);
     const protectedSummaries = boundedResourceSummaries(protectedActiveRunners);
     const selectedSummaries = boundedResourceSummaries(selected);
     const boundedOutcomes = boundedJsonRecords(outcomes);
@@ -315,6 +315,8 @@ export async function reconcileStalePendingRunners(input: { store: AgentRunStore
       nonTerminalRunnerPodCountBefore: before.nonTerminalRunnerPodCount,
       orphanNonTerminalRunnerPodCountBefore: before.orphanNonTerminalRunnerPodCount,
       inactiveCandidateCount: resources.filter((item) => !item.protectedActive).length,
+      inactiveCandidates: inactiveCandidateSummaries.items,
+      inactiveCandidateOmittedCount: inactiveCandidateSummaries.omittedCount,
       stalePendingCandidateCount: stalePendingCandidates.length,
       stalePendingCandidates: candidateSummaries.items,
       stalePendingCandidateOmittedCount: candidateSummaries.omittedCount,
@@ -337,7 +339,7 @@ export async function reconcileStalePendingRunners(input: { store: AgentRunStore
       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",
+      reason: postObservationError ? "inactive-runner-post-observation-failed" : selected.length === 0 ? "no-inactive-runner-candidate" : failedRunnerCount > 0 ? "inactive-runner-reconcile-partial" : "inactive-runner-reconciled",
       cleanupPolicy: cleanupPolicySummary(),
       valuesPrinted: false,
     } satisfies RunnerStalePendingReconcileSummary;
@@ -508,16 +510,12 @@ async function activityFor(store: AgentRunStore, input: {
     return protectedActivity(input, "db-progress-facts-unavailable", lastActiveAtMs, null);
   }
   const dbFactsFingerprint = databaseFactsFingerprint(run, command, commands, events);
-  const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs));
-  if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs, dbFactsFingerprint);
-  const activePodReason = activePodProtectionReason(input.podObservations);
-  if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs, dbFactsFingerprint);
   if (isTerminalRunStatus(run.status)) {
     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);
+    const claimDrift = run.claimedBy !== null || run.leaseExpiresAt !== null;
     return candidateActivity(
       input,
       input.terminal ? "terminal" : "idle",
@@ -527,6 +525,10 @@ async function activityFor(store: AgentRunStore, input: {
       claimDrift ? "release-terminal-command-claim" : "none",
     );
   }
+  const freshClaim = Boolean(run.claimedBy && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs));
+  if (freshClaim) return protectedActivity(input, "fresh-active-run", lastActiveAtMs, dbFactsFingerprint);
+  const activePodReason = activePodProtectionReason(input.podObservations);
+  if (activePodReason) return protectedActivity(input, activePodReason, lastActiveAtMs, dbFactsFingerprint);
   if (input.terminal) return candidateActivity(input, "terminal", "terminal-runner-resource", lastActiveAtMs, dbFactsFingerprint);
 
   const podBlockReason = stalePendingPodBlockReason(input.podObservations);

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

@@ -123,6 +123,7 @@ const selfTest: SelfTestCase = async (context) => {
     await assertEventOnlyFailedMountCandidate({ fixturePath, statePath, kubectlCommand });
     await assertTerminalClaimDriftCandidate({ fixturePath, statePath, kubectlCommand });
     await assertTerminalCommandClaimDriftCandidate({ fixturePath, statePath, kubectlCommand });
+    await assertTerminalFactsOverrideHistoricalActivity({ fixturePath, statePath, kubectlCommand });
     await assertTerminalClaimRepairCasConflictFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertProtectedMatrix({ fixturePath, statePath, kubectlCommand });
     await assertPendingToRunningRaceFailsClosed({ fixturePath, statePath, kubectlCommand });
@@ -137,6 +138,7 @@ const selfTest: SelfTestCase = async (context) => {
     await assertMemoryDbFenceRejectsConcurrentClaim();
     await assertRunnerCreateFenceSerializesNamespace();
     await assertPeriodicReconcilerCleansAtCapacity({ fixturePath, statePath, kubectlCommand });
+    await assertPeriodicReconcilerCleansTerminalRunners({ fixturePath, statePath, kubectlCommand });
     await assertPeriodicReconcilerProtectsUnsafeRunners({ fixturePath, statePath, kubectlCommand });
     await assertPeriodicReconcilerCasConflictFailsClosed({ fixturePath, statePath, kubectlCommand });
     await assertConcurrentPeriodicReconcileIsSerialized({ fixturePath, statePath, kubectlCommand });
@@ -167,9 +169,9 @@ const selfTest: SelfTestCase = async (context) => {
       "container-started-pending-protected",
       "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-facts-override-historical-claim-heartbeat-and-running-pod",
       "terminal-claim-repair-cas-conflict-fails-closed",
       "terminal-runner-pod-completed-is-not-active",
       "selected-runner-race-fails-closed",
@@ -184,6 +186,7 @@ const selfTest: SelfTestCase = async (context) => {
       "memory-db-fence-rejects-concurrent-claim",
       "namespace-runner-create-fence-prevents-capacity-overshoot",
       "periodic-reconciler-cleans-stale-pending-at-capacity",
+      "periodic-reconciler-cleans-terminal-inactive-runners",
       "periodic-reconciler-protects-unsafe-runners",
       "periodic-reconciler-cas-conflict-fails-closed",
       "periodic-reconciler-namespace-fence-prevents-duplicate-terminalization",
@@ -268,23 +271,6 @@ async function assertProtectedMatrix(input: { fixturePath: string; statePath: st
   await assertProtected(input, [base], [], "pod-facts-unavailable");
   await assertProtected(input, [base], [runnerPod(base, 100, "pending")], "pod-waiting-event-facts-unavailable", base, { eventsError: true });
   await assertProtected(input, [base], [runnerPod(base, 100, "pending"), runnerPod(base, 101, "running", 100)], "runner-pod-started-running-or-ready");
-  const terminalRunning = pendingFact(105, oldAt);
-  terminalRunning.run.status = "completed";
-  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");
@@ -295,6 +281,31 @@ async function assertProtectedMatrix(input: { fixturePath: string; statePath: st
   await assertProtected(input, [mismatchedUserMessage], [runnerPod(mismatchedUserMessage, 106, "failed-config")], "business-progress-observed-or-unverifiable");
 }
 
+async function assertTerminalFactsOverrideHistoricalActivity(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const oldAt = new Date(Date.now() - 60 * 60_000).toISOString();
+  const terminalRun = pendingFact(105, oldAt, "claimed");
+  terminalRun.run.status = "completed";
+  terminalRun.run.terminalStatus = "completed";
+  terminalRun.command.state = "completed";
+  await writeFixture(input, { jobs: [runnerJob(terminalRun, 105)], pods: [runnerPod(terminalRun, 105, "running")], events: [] });
+  const runSummary = await enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([terminalRun]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 0 });
+  const runCandidate = firstSummary(runSummary, "inactiveCandidates");
+  assert.equal(runSummary.protectedActiveRunnerCount, 0);
+  assert.equal(runCandidate.candidateKind, "terminal");
+  assert.equal(runCandidate.classificationReason, "terminal-run-stale-claim");
+  assert.equal(runCandidate.repairKind, "release-terminal-run-claim");
+
+  const terminalCommand = pendingFact(107, oldAt, "claimed");
+  terminalCommand.command.state = "completed";
+  await writeFixture(input, { jobs: [runnerJob(terminalCommand, 107)], pods: [runnerPod(terminalCommand, 107, "running")], events: [] });
+  const commandSummary = await enforceRunnerRetentionBeforeCreate({ store: new RetentionFactStore([terminalCommand]), options: retentionOptions(input.kubectlCommand, 1), incomingRunnerCount: 0 });
+  const commandCandidate = firstSummary(commandSummary, "inactiveCandidates");
+  assert.equal(commandSummary.protectedActiveRunnerCount, 0);
+  assert.equal(commandCandidate.candidateKind, "idle");
+  assert.equal(commandCandidate.classificationReason, "terminal-command-stale-claim");
+  assert.equal(commandCandidate.repairKind, "release-terminal-command-claim");
+}
+
 async function assertLegacyNoUserMessageCandidate(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
   const fact = pendingFact(89, new Date(Date.now() - 60 * 60_000).toISOString());
   fact.events = fact.events
@@ -668,7 +679,7 @@ async function assertPeriodicReconcilerCleansAtCapacity(input: { fixturePath: st
   assert.equal(retention.terminalizedCommandCount, 3);
   assert.equal(retention.deletedRunnerJobCount, 3);
   assert.equal(retention.failedRunnerCount, 0);
-  assert.equal(retention.reason, "stale-pending-reconciled");
+  assert.equal(retention.reason, "inactive-runner-reconciled");
   assert.equal(store.terminalizationCount, 3);
   for (const fact of facts.slice(0, 3)) {
     assert.equal(store.getRun(fact.run.id).status, "failed");
@@ -684,6 +695,37 @@ async function assertPeriodicReconcilerCleansAtCapacity(input: { fixturePath: st
   );
 }
 
+async function assertPeriodicReconcilerCleansTerminalRunners(input: { fixturePath: string; statePath: string; kubectlCommand: string }): Promise<void> {
+  const oldAt = new Date(Date.now() - 60 * 60_000).toISOString();
+  const facts = [360, 361].map((index) => {
+    const fact = pendingFact(index, oldAt, "claimed");
+    fact.run.status = "completed";
+    fact.run.terminalStatus = "completed";
+    fact.command.state = "completed";
+    return fact;
+  });
+  await writeFixture(input, {
+    jobs: facts.map((fact, index) => runnerJob(fact, index + 360)),
+    pods: facts.map((fact, index) => runnerPod(fact, index + 360, "running")),
+  });
+  const store = new RetentionFactStore(facts);
+  const summary = await reconcileRunnerJobsOnce({
+    store,
+    namespace: "agentrun-nc01-v02",
+    kubectlCommand: input.kubectlCommand,
+    retention: retentionOptions(input.kubectlCommand, 20),
+    limit: 20,
+  });
+  const retention = summary.stalePendingRetention as JsonRecord;
+  assert.equal(retention.inactiveCandidateCount, 2);
+  assert.equal(retention.stalePendingCandidateCount, 0);
+  assert.equal(retention.selectedRunnerCount, 2);
+  assert.equal(retention.deletedRunnerJobCount, 2);
+  assert.equal(retention.reason, "inactive-runner-reconciled");
+  assert.equal(store.getRun(facts[0]!.run.id).claimedBy, null);
+  assert.equal(store.getRun(facts[1]!.run.id).claimedBy, null);
+}
+
 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();