diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 200abff..1d4a3e5 100644 --- a/src/mgr/postgres-store.ts +++ b/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 { + if (!isTerminalRunStatus(run.status)) return run; + return await this.releaseRunnerClaim(client, run); + } + + private async releaseRunnerClaim(client: PoolClient, run: RunRecord): Promise { + 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(fn: (client: PoolClient) => Promise): Promise { const client = await this.pool.connect(); try { diff --git a/src/mgr/runner-retention.ts b/src/mgr/runner-retention.ts index 9e9a30b..e2b6a53 100644 --- a/src/mgr/runner-retention.ts +++ b/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; +type ActivityFields = Pick; 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 { @@ -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 { @@ -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, }; } diff --git a/src/mgr/store.ts b/src/mgr/store.ts index e09649c..5774cd2 100644 --- a/src/mgr/store.ts +++ b/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): 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))) { diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index 1cd13ba..a2d7e0e 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/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 { + 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 { + 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 { + 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 { 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 { let releaseFence = (): void => {}; const entered = new Promise((resolve) => { enterFence = resolve; }); const release = new Promise((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; })); diff --git a/src/selftest/cases/23-runner-terminal-lifecycle.ts b/src/selftest/cases/23-runner-terminal-lifecycle.ts new file mode 100644 index 0000000..129d67f --- /dev/null +++ b/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; diff --git a/src/selftest/integration/postgres-runner-retention-fence.ts b/src/selftest/integration/postgres-runner-retention-fence.ts index cb70959..73d91ba 100644 --- a/src/selftest/integration/postgres-runner-retention-fence.ts +++ b/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 { @@ -80,7 +85,7 @@ async function assertRetentionFenceBlocksClaim(): Promise { let releaseFence = (): void => {}; const entered = new Promise((resolve) => { fenceEntered = resolve; }); const release = new Promise((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 { 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 { + 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 event.payload.retentionFence === true).length, 0); } +async function assertTerminalLifecycleReleasesClaim(): Promise { + 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 { + 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 { + 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 { + 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 { + 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 { + 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 { await new Promise((resolve) => setTimeout(resolve, ms)); }