From 63d33c07f8717aac42a677699f4af55c8895794c Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 20:04:46 -0500 Subject: [PATCH] Checkpoint residual recovery deadline rechecks and owning producers Co-Authored-By: Paperclip --- server/src/services/agent-configuration.ts | 2 ++ .../agent-instruction-working-copies.ts | 5 ++- server/src/services/agent-lifecycle.ts | 2 ++ .../environment-cleanup-notifications.ts | 13 +++++++ .../execution-control-notifications.ts | 6 ++-- server/src/services/heartbeat-run-events.ts | 6 +++- server/src/services/heartbeat.ts | 4 +++ server/src/services/heartbeat/agent-status.ts | 19 +++++++++- .../services/heartbeat/live-recovery-work.ts | 35 +++++++++++++------ server/src/services/heartbeat/recovery.ts | 34 ++++++++++++++++-- .../src/services/legacy-controller-lease.ts | 14 ++++++-- .../legacy-shutdown-workspace-settlement.ts | 13 +++++-- .../live-run-recovery-notifications.ts | 5 +++ .../native-finalization-notifications.ts | 4 +-- .../src/services/native-local-process-stop.ts | 2 ++ .../native-runtime/native-execution-lease.ts | 2 ++ .../native-finalization-reconciler.ts | 4 +++ .../native-runtime/native-question-bridge.ts | 2 ++ .../native-runtime/native-restart-recovery.ts | 3 ++ .../native-runner-checkpoint.ts | 2 ++ 20 files changed, 151 insertions(+), 26 deletions(-) diff --git a/server/src/services/agent-configuration.ts b/server/src/services/agent-configuration.ts index 2d30da27cb7..3b23d901eef 100644 --- a/server/src/services/agent-configuration.ts +++ b/server/src/services/agent-configuration.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionPinChange } from "./skill-version-retention-notifications.js"; @@ -85,6 +86,7 @@ export async function prepareAgentConfiguration(db: Db, existing: AgentConfigura existing.adapterConfig, ); } + await notifyStrandedAgentStatus(db, { companyId: existing.companyId, agentId: existing.id }); await notifyRunDispatch(db, { companyId: existing.companyId }); await notifyDeferredExecutionEligibility(db, { companyId: existing.companyId }); if (normalizedPatch.runtimeConfig !== undefined || normalizedPatch.reportsTo !== undefined) { diff --git a/server/src/services/agent-instruction-working-copies.ts b/server/src/services/agent-instruction-working-copies.ts index e0309bc8792..6cb0aebb3d4 100644 --- a/server/src/services/agent-instruction-working-copies.ts +++ b/server/src/services/agent-instruction-working-copies.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyInstructionCollection, type InstructionCollectionScope } from "./instruction-collection-notifications.js"; import { notifyDirectoryCleanup, type DirectoryCleanupScope } from "./directory-cleanup-notifications.js"; import { notifyInstructionCommit, type InstructionCommitScope } from "./instruction-commit-notifications.js"; @@ -101,6 +102,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen const signalsCleanup = directoryCleanupEntry(row) !== null || directoryCleanupEntry({ ...row, ...values }) !== null; const signalsCommit = row.state === "pending_commit" || values.state === "pending_commit"; const write = async (tx: Db) => { + await notifyStrandedAgentStatus(tx, { companyId: row.companyId, agentId: row.agentId }); if (signalsCommit) await notifyInstructionCommit(tx, { companyId: row.companyId, runId: row.runId }); if (signalsCleanup) await notifyDirectoryCleanup(tx, { companyId: row.companyId, runId: row.runId }); @@ -112,7 +114,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen )).returning(); return updated; }; - return signalsCommit || signalsCleanup || signalsCollection ? db.transaction(tx => write(tx as unknown as Db)) : write(db); + return db.transaction(tx => write(tx as unknown as Db)); } async function patch(row: Copy, values: Partial) { const updated = await tryPatch(row, values); @@ -204,6 +206,7 @@ export function agentInstructionWorkingCopyService(db: Db, options: { environmen if (!existing) { await db.transaction(async tx => { await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId }); + await notifyStrandedAgentStatus(tx, { companyId: input.companyId, agentId: input.agentId }); await tx.insert(copies).values({ runId: input.runId, companyId: input.companyId, agentId: input.agentId, responsibleUserId: bound.onBehalfOfUserId!, entryFile, diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 8bb46882256..9302e2796c7 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; @@ -41,6 +42,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), async beforeStatusChange(tx, companyId, agentId, before, after) { if (before !== after) { + await notifyStrandedAgentStatus(tx, { companyId, agentId }); await notifyRunDispatch(tx, { companyId }); await notifyDeferredExecutionEligibility(tx, { companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts index 728f05b9303..bd26350e3ab 100644 --- a/server/src/services/environment-cleanup-notifications.ts +++ b/server/src/services/environment-cleanup-notifications.ts @@ -1,8 +1,21 @@ +import { and, eq, sql } from "drizzle-orm"; +import { environmentLeases } from "@paperclipai/db"; +import { notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement, type NativeEnvironmentSettlementScope } from "./native-environment-settlement-notifications.js"; import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export async function notifyEnvironmentCleanup(transaction: object, scope?: NativeEnvironmentSettlementScope): Promise { + await notifyLiveResourceRecovery(transaction, scope); + if (scope?.runId && typeof (transaction as Db).select === "function") { + const peers = await (transaction as Db).select({ id: environmentLeases.id, companyId: environmentLeases.companyId, + runId: environmentLeases.heartbeatRunId }).from(environmentLeases).where(and(eq(environmentLeases.status, "active"), + sql`exists (select 1 from environment_leases changed where changed.company_id = ${scope.companyId} + and changed.heartbeat_run_id = ${scope.runId} and changed.provider_lease_id is not null + and changed.provider = ${environmentLeases.provider} and changed.provider_lease_id = ${environmentLeases.providerLeaseId})`)); + for (const peer of peers) await notifyDeliveryWork(transaction, DELIVERY_QUEUES.orphanActiveLeases, + JSON.stringify({ companyId: peer.companyId, leaseId: peer.id })); + } await notifyNativeEnvironmentSettlement(transaction, scope); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 615c6452327..b79e0a7b24e 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,4 +1,4 @@ -import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; +import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import type { Db } from "@paperclipai/db"; @@ -10,7 +10,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut if (scope) { await notifyRunDispatch(transaction, scope); await notifyDeferredExecutionEligibility(transaction, scope); - await notifyActiveRunSilence(transaction, scope); + await notifyLiveRunRecovery(transaction, scope); } for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { @@ -28,7 +28,7 @@ export function writeExecutionControl( await notifyRunDispatch(transaction, scope); if (scope) { await notifyDeferredExecutionEligibility(transaction, scope); - await notifyActiveRunSilence(transaction, scope); + await notifyLiveRunRecovery(transaction, scope); } await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index ebce78a8266..a755062048e 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js"; import { readNativeEventEvidenceOutcome } from "./native-runtime/evidence-classifier.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; @@ -74,10 +75,11 @@ export async function appendHeartbeatRunEvent( db: Db, input: AppendHeartbeatRunEventInput, ): Promise { + const projectsRecovery = ["native.process_start_requested", "native.process_identity_recorded", "native.local_process_stopped"].includes(input.eventType); const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null; const projectsProgress = isSafeNativeChatProgressEventType(input.eventType); const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent); - if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment) { + if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRecovery) { const allocated = db.$with("allocated_run_event", {}).as(sql` update ${heartbeatRuns} set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1, @@ -117,6 +119,8 @@ export async function appendHeartbeatRunEvent( throw new Error("heartbeat_run_event_binding_mismatch"); } + if (projectsRecovery) await notifyLiveRunRecovery(tx, input); + if (input.retryExhaustion && !input.nativeSource) { // The run lock also serializes concurrent recovery checks across server // instances. Reusing the receipt must not allocate a sequence or publish diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 57328de4f7a..f04c192efb3 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1026,6 +1026,7 @@ export function heartbeatService( }); const { + registerResidualRecoveryWork: registerResidualRecoveryOwner, sweepPendingCleanupLeases, markNativeOwnershipUnverified, prepareHotRestartShutdown, @@ -4245,6 +4246,9 @@ export function heartbeatService( computeTaskDrain, applyTaskDrain, + registerResidualRecoveryWork: (coordinator: Parameters[0], canRun: () => boolean) => + registerResidualRecoveryOwner(coordinator, () => canRun() && readTaskDrain(new Date()) === null, + async () => !(await getSchedulingSuppression(true)).suppressed), registerActiveRunSilenceWork: (coordinator: Parameters[0], canRun: () => boolean) => recovery.registerActiveRunSilenceWork(coordinator, () => canRun() && readTaskDrain(new Date()) === null, async scope => recovery.scanSilentActiveRuns({ ...scope, issueCreatedAtGte: await getWorktreeExecutionCutoff() }), diff --git a/server/src/services/heartbeat/agent-status.ts b/server/src/services/heartbeat/agent-status.ts index 2957471926a..a53cdb5cd52 100644 --- a/server/src/services/heartbeat/agent-status.ts +++ b/server/src/services/heartbeat/agent-status.ts @@ -1,3 +1,4 @@ +import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { agents, heartbeatRuns, type Db } from "@paperclipai/db"; import { and, asc, desc, eq, lte, sql } from "drizzle-orm"; @@ -91,6 +92,7 @@ export function createHeartbeatAgentStatus(db: Db) { )).limit(1); const nextStatus = running ? "running" : settledStatus(outcome, options?.keepIdleOnFailure); const now = new Date(); + await notifyStrandedAgentStatus(tx, { companyId: existing.companyId, agentId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); const [updated] = await tx.update(agents).set({ status: nextStatus, @@ -109,7 +111,7 @@ export function createHeartbeatAgentStatus(db: Db) { publishStatus(result.updated, outcome); } - async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet, scope: { companyId?: string; agentId?: string } = {}) { + async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet, scope: { companyId?: string; agentId?: string; onDeferredDeadline?: (at: number) => void } = {}) { const now = new Date(); const cutoff = new Date(now.getTime() - RECONCILIATION_GRACE_MS); const candidates = await db.transaction(async tx => { @@ -158,6 +160,21 @@ export function createHeartbeatAgentStatus(db: Db) { for (const publication of publications) publishActivity(publication); publishStatus(updated, "reconciled"); } + if (scope.companyId && scope.agentId && reconciled.length === 0 && scope.onDeferredDeadline) { + const rows = await db.execute(sql`select min(deadline) as deadline from ( + select r.controller_lease_expires_at as deadline from heartbeat_runs r + where r.company_id = ${scope.companyId} and r.agent_id = ${scope.agentId} + union all + select f.lease_expires_at as deadline from native_run_finalizations f + inner join heartbeat_runs r on r.id = f.run_id and r.company_id = f.company_id + where r.company_id = ${scope.companyId} and r.agent_id = ${scope.agentId} + ) authority where deadline > ${now.toISOString()}::timestamptz`); + const deadline = rows[0]?.deadline; + if (deadline) { + const at = deadline instanceof Date ? deadline.getTime() : Date.parse(String(deadline)); + if (Number.isFinite(at) && at > Date.now()) scope.onDeferredDeadline(at); + } + } return reconciled; } diff --git a/server/src/services/heartbeat/live-recovery-work.ts b/server/src/services/heartbeat/live-recovery-work.ts index eb3e41204b1..d2d9ab78a5b 100644 --- a/server/src/services/heartbeat/live-recovery-work.ts +++ b/server/src/services/heartbeat/live-recovery-work.ts @@ -25,16 +25,7 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType(); let enabled = false; - return registerScopedDeadlineWork(coordinator, { - queue, - reconstructionIntervalMs: 30_000, - beforePass: async () => { enabled = await owner.enabled(); }, - decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint), - matches: (entry, scope) => !scope || entry.companyId === scope.companyId && ( - kind === "agent" ? !scope.agentId || entry.id === scope.agentId - : kind === "orphanLease" || kind === "shutdown" ? (!scope.leaseId || entry.id === scope.leaseId) && (!scope.runId || entry.runId === scope.runId) - : !scope.runId || entry.id === scope.runId), - load: async scope => { + async function load(scope?: Scope) { const key = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, leaseId: scope.leaseId, agentId: scope.agentId }) : "*"; if (!enabled) { cursors.delete(key); return []; } @@ -89,9 +80,26 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType(coordinator, { + queue, + reconstructionIntervalMs: 30_000, + beforePass: async () => { enabled = await owner.enabled(); }, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry, scope) => !scope || entry.companyId === scope.companyId && ( + kind === "agent" ? !scope.agentId || entry.id === scope.agentId + : kind === "orphanLease" || kind === "shutdown" ? (!scope.leaseId || entry.id === scope.leaseId) && (!scope.runId || entry.runId === scope.runId) + : !scope.runId || entry.id === scope.runId), + load, check: async entry => { if (!enabled) return null; + const currentScope = { companyId: entry.companyId, + ...(kind === "agent" ? { agentId: entry.id } : kind === "orphanLease" || kind === "shutdown" + ? { leaseId: entry.id } : { runId: entry.id }) }; + const loaded = await load(currentScope); + const current = (Array.isArray(loaded) ? loaded : loaded.rows).find(row => row.id === entry.id); + if (!current) return null; + if (current.dueAt > Date.now()) return current; let next: number | null = null; const errors: unknown[] = []; const scope: CheckScope = { companyId: entry.companyId, @@ -102,6 +110,11 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType row.id === entry.id); + if (remaining && remaining.dueAt > Date.now()) next = next === null ? remaining.dueAt : Math.min(next, remaining.dueAt); + } return next === null ? null : { ...entry, dueAt: next }; }, }, canRun); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 141d0f17755..75afd5d881c 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,3 +1,4 @@ +import { registerLiveRecoveryWork } from "./live-recovery-work.js"; import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../native-runtime/native-environment-settlement-work.js"; import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js"; import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js"; @@ -18,7 +19,7 @@ import { hasStopOnlyCleanup, settleStopOnlyCleanup, } from "../sandbox-stop-and-retain.js"; -import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; +import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, releaseSettledNativeRunProcess, retryPendingSettledNativeRunProcesses } from "../native-local-process-stop.js"; import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; import { legacyControllerBootId, @@ -2010,6 +2011,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD eq(heartbeatRuns.runtimeMode, "native"), eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId), + sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`, ), ); const claimableNativeRunIds = new Set(); @@ -2260,6 +2262,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD nativeControllerLeaseExpiresAt, nativeWorkspaceFinalizationPending, } of activeRuns) { + if (parseObject(run.contextSnapshot).nativeQuestionCancellation) continue; // Native recovery owns these dispositions. Its writes happened after this // inventory was read; do not reap or overwrite them using the old row. if (remoteRecoveryRunIds.has(run.id)) continue; @@ -2329,10 +2332,16 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD processPidAlive: nativeProcessPidAlive, processGroupAlive: nativeProcessGroupAlive, }); + if (nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now) + scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime()); continue; } if (resumedRunIds.has(run.id)) continue; - if (locallyTracked) continue; + if (locallyTracked) { + if (!activeRunExecutions.has(run.id) && !runningProcesses.has(run.id) && nativeControllerLeaseExpiresAt) + scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime()); + continue; + } const legacyLease = await liveLegacyControllerExpiresAt(db, run); if (legacyLease) { scope.onDeferredDeadline?.(legacyLease.getTime()); continue; } @@ -2674,7 +2683,28 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD return { reaped: reaped.length, runIds: reaped }; } + function registerResidualRecoveryWork(coordinator: Parameters[1], + canRun: () => boolean, enabled: () => Promise) { + return registerLiveRecoveryWork(db, coordinator, { + native: scope => resumeRetryableNativeProcessSessions(scope), + question: recoverQuestionCancellationRequests, + active: scope => recoverActiveExecutions(scope), + settled: async scope => { + if (!scope.runId || activeRunExecutions.has(scope.runId) || runningProcesses.has(scope.runId) + || adapterExecutionControls.has(scope.runId)) return; + await releaseSettledNativeRunProcess(db, scope.runId, scope); + }, + orphanLease: scope => sweepOrphanedActiveLeases({ ...scope, backoffMs: 5 * 60_000 }), + shutdown: scope => sweepLegacyShutdownWorkspaceSettlements(db, new Date(), scope), + agent: scope => createHeartbeatAgentStatus(db).reconcileStrandedAgentStatuses(() => new Set([ + ...activeRunExecutions, ...runningProcesses.keys(), ...adapterExecutionControls.keys(), + ]), scope), + enabled, + }, canRun); + } + return { + registerResidualRecoveryWork, registerRetainedNativeSessionCleanupWork: retainedNativeCleanup.register, registerNativeWorkspaceFinalizationWork: nativeWorkspaceFinalization.register, registerNativeEnvironmentSettlementWork: nativeEnvironmentSettlement.register, diff --git a/server/src/services/legacy-controller-lease.ts b/server/src/services/legacy-controller-lease.ts index 22fa078f459..95ebc8db584 100644 --- a/server/src/services/legacy-controller-lease.ts +++ b/server/src/services/legacy-controller-lease.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, gt, lte, sql } from "drizzle-orm"; import { heartbeatRuns, livenessDb, type Db } from "@paperclipai/db"; @@ -40,7 +41,9 @@ export async function renewLegacyControllerLease( run: Pick, stage?: "dispatching", ): Promise { - const [renewed] = await db.update(heartbeatRuns).set({ + const [renewed] = await db.transaction(async tx => { + await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id }); + return tx.update(heartbeatRuns).set({ controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`, ...(stage ? { executionStage: stage } : {}), }).where(and( @@ -49,6 +52,7 @@ export async function renewLegacyControllerLease( eq(heartbeatRuns.controllerBootId, legacyControllerBootId), gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), )).returning({ id: heartbeatRuns.id }); + }); return Boolean(renewed); } @@ -105,19 +109,23 @@ export async function revokeExpiredLegacyControllerWithState( if (run.runtimeMode === "native" || !run.controllerBootId) { return { executionStage: run.executionStage, processPid: run.processPid, processGroupId: run.processGroupId }; } - const [revoked] = await db.update(heartbeatRuns).set({ + const controllerBootId = run.controllerBootId; + const [revoked] = await db.transaction(async tx => { + await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id }); + return tx.update(heartbeatRuns).set({ controllerBootId: randomUUID(), controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`, }).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "running"), - eq(heartbeatRuns.controllerBootId, run.controllerBootId), + eq(heartbeatRuns.controllerBootId, controllerBootId), lte(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), )).returning({ executionStage: heartbeatRuns.executionStage, processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, }); + }); return revoked ?? null; } diff --git a/server/src/services/legacy-shutdown-workspace-settlement.ts b/server/src/services/legacy-shutdown-workspace-settlement.ts index 8c85cd3f14b..4a3576f997d 100644 --- a/server/src/services/legacy-shutdown-workspace-settlement.ts +++ b/server/src/services/legacy-shutdown-workspace-settlement.ts @@ -1,3 +1,4 @@ +import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { and, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, issues, type Db } from "@paperclipai/db"; import { adapterExecutionControls } from "./adapter-execution-control.js"; @@ -33,6 +34,7 @@ export function maintainLegacyShutdownWorkspaceSettlement(db: Db, run: Run) { pending = db.transaction(async tx => { await tx.execute(sql`SET LOCAL lock_timeout = '1s'`); await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); await tx.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, '{legacyShutdownWorkspaceSettlement,deadlineAt}', ${JSON.stringify(new Date(Date.now() + SETTLEMENT_LEASE_MS).toISOString())}::jsonb)` }) .where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), @@ -86,23 +88,29 @@ export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new /** Commit the fence while the run is still live, before exposing interruption. */ export async function beginLegacyShutdownWorkspaceSettlement(db: Db, run: Run, deadlineAt: Date) { - await db.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || + await db.transaction(async tx => { + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); + await tx.update(environmentLeases).set({ metadata: sql`coalesce(${environmentLeases.metadata}, '{}'::jsonb) || ${JSON.stringify({ legacyShutdownWorkspaceSettlement: { schema: SCHEMA, state: "pending", runId: run.id, controllerBootId: legacyControllerBootId, deadlineAt: deadlineAt.toISOString() } })}::jsonb`, }).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.status, "active"), ne(environmentLeases.provider, "local"), or(sql`${environmentLeases.metadata}->>'driver' = 'sandbox'`, sql`${environmentLeases.metadata}->>'sandboxProviderPlugin' = 'true'`))); + }); } /** The owner joins settlement, or records exact retention before timeout cleanup. */ export async function finishLegacyShutdownWorkspaceSettlement(db: Db, run: Run, state: "settled" | "expired" = "settled") { - await db.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, + await db.transaction(async tx => { + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); + await tx.update(environmentLeases).set({ metadata: sql`jsonb_set(${environmentLeases.metadata}, '{legacyShutdownWorkspaceSettlement,state}', ${JSON.stringify(state)}::jsonb)` }).where(and( eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'schema' = ${SCHEMA}`, sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'controllerBootId' = ${legacyControllerBootId}`, sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`, )); + }); } /** A restart or another cleanup worker must never infer saved files from timeout. */ @@ -140,6 +148,7 @@ export async function allowLegacyShutdownWorkspaceCleanup(db: Db, lease: Lease, if (!Number.isFinite(deadline) || deadline > now.getTime()) return false; // A killed container cannot certify export. Preserve source and hold in // this same transaction before any worker may perform provider cleanup. + await notifyEnvironmentCleanup(tx, { companyId: run.companyId, runId: run.id }); if (current.status !== "active") { const { hasConfirmedSandboxStopAndRetain, readStopOnlyCleanup } = await import("./sandbox-stop-and-retain.js"); if (!(current.status === "released" && hasConfirmedSandboxStopAndRetain(current)) diff --git a/server/src/services/live-run-recovery-notifications.ts b/server/src/services/live-run-recovery-notifications.ts index 10334f6776c..b270e3b9564 100644 --- a/server/src/services/live-run-recovery-notifications.ts +++ b/server/src/services/live-run-recovery-notifications.ts @@ -23,3 +23,8 @@ export async function notifyLiveResourceRecovery(tx: object, scope?: LiveResourc DELIVERY_QUEUES.strandedAgentStatus, DELIVERY_QUEUES.settledNativeProcesses]) await notifyDeliveryWork(tx, queue, hint); } + +export async function notifyStrandedAgentStatus(tx: object, scope: { companyId: string; agentId: string }) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAgentStatus, + JSON.stringify({ companyId: scope.companyId, agentId: scope.agentId })); +} diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index a38aae46c0b..2427ca0fd17 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,4 +1,4 @@ -import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; +import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; @@ -16,7 +16,7 @@ export type NativeFinalizationWorkScope = { companyId: string; runId?: string; i export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { if (scope.issueId || scope.runId) { - await notifyActiveRunSilence(transaction, scope); + await notifyLiveRunRecovery(transaction, scope); await notifyNativeEnvironmentSettlement(transaction, scope); await notifyExternalAgentSettlement(transaction, scope); await notifyExternalConversationSettlement(transaction, scope); diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index b26a29e7add..e30c18e9cdc 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; @@ -178,6 +179,7 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string, opti return "pending"; } } + await notifyLiveRunRecovery(tx, { companyId: run.companyId, runId: run.id, agentId: run.agentId }); await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id }); await tx.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null, updatedAt: new Date(), diff --git a/server/src/services/native-runtime/native-execution-lease.ts b/server/src/services/native-runtime/native-execution-lease.ts index 645cdb24fdf..c91803da33c 100644 --- a/server/src/services/native-runtime/native-execution-lease.ts +++ b/server/src/services/native-runtime/native-execution-lease.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import type { Db } from "@paperclipai/db"; import { livenessDb, nativeRunFinalizations } from "@paperclipai/db"; import { and, eq, gt, sql } from "drizzle-orm"; @@ -53,6 +54,7 @@ async function renewNativeSessionExecutionLeaseDeadline( // but conservatively include UPDATE/commit response time without assuming // that this server's wall clock agrees with the database's clock. confirmedDeadline = performance.now() + leaseTtlMs; + await notifyLiveRunRecovery(tx, { companyId: input.companyId, runId: input.runId }); return tx.update(nativeRunFinalizations) .set({ leaseExpiresAt: sql`clock_timestamp() + (${leaseTtlMs} * interval '1 millisecond')`, diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index af828edfb98..d24988d291c 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; @@ -244,6 +245,7 @@ export async function claimNativeSessionResumptions(input: { ) { const failureCode = originalFailureCode; await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, row.run.companyId); + await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); await tx.update(nativeRunFinalizations).set({ phase: "terminal_failure", leaseOwner: null, @@ -312,6 +314,7 @@ export async function claimNativeSessionResumptions(input: { } const leaseExpiresAt = new Date(now.getTime() + 20 * 60_000); + await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); const updated = await tx.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner, @@ -362,6 +365,7 @@ export async function claimNativeSessionResumptions(input: { if (claimed) claims.push({ runId: candidate.runId, leaseOwner }); else await input.db.transaction(async tx => { await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`); + await notifyLiveRunRecovery(tx, { companyId: candidate.companyId, runId: candidate.runId }); await tx.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(now.getTime() + 30_000), updatedAt: now, diff --git a/server/src/services/native-runtime/native-question-bridge.ts b/server/src/services/native-runtime/native-question-bridge.ts index f6ae8d7ae82..1ebc09d0e0f 100644 --- a/server/src/services/native-runtime/native-question-bridge.ts +++ b/server/src/services/native-runtime/native-question-bridge.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { and, eq, inArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -348,6 +349,7 @@ export async function requestNativeQuestionRunCancellation( ...cause, requestedAt: new Date().toISOString(), }); + await notifyLiveRunRecovery(db, { companyId: interaction.companyId, runId: run.id }); return db.update(heartbeatRuns).set({ contextSnapshot: sql`jsonb_set( case diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index b54f6019f3b..03649cf852b 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -649,6 +650,7 @@ export async function claimNativeRestartRecoveries(input: { hasProviderEvidence: false, stderrTail: row.run.stderrExcerpt, }); + await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); await tx .update(nativeRunFinalizations) .set({ @@ -1020,6 +1022,7 @@ export async function claimNativeRestartRecoveries(input: { recycledProviderPids: providerProcesses.recycledPids, stderrTail: row.run.stderrExcerpt, }); + await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); const claimed = await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index 3cdef6040e2..455abdda25a 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -1,3 +1,4 @@ +import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm"; import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db"; import { inspectDurablePrpCheckpointHeader, decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; @@ -75,6 +76,7 @@ export function nativeRunnerCheckpointWriter(db: Db, owner: Owner) { eq(heartbeatRuns.nativeSessionId, checkpoint.predecessor.normalizedSessionId))); if (!prior) throw new Error("native_runner_checkpoint_binding_mismatch"); } + await notifyLiveRunRecovery(tx, { companyId: owner.companyId, runId: owner.runId }); if (checkpoint.reset) await tx.delete(nativeRunnerCheckpoints).where(and( eq(nativeRunnerCheckpoints.companyId, owner.companyId), eq(nativeRunnerCheckpoints.runId, owner.runId))); else if (checkpoint.removedRecords.length) await tx.delete(nativeRunnerCheckpoints).where(and(