diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index ae3ed85..4ac3a33 100644 --- a/src/mgr/runner-retention.ts +++ b/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); diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index baef697..e2c559f 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/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 { + 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 { 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 { + 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 { const oldAt = new Date(Date.now() - 60 * 60_000).toISOString(); const freshAt = new Date(Date.now() - 60_000).toISOString();