From 30b927635411d1c92b61fb1f21be181a7ac8a4f3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:32:47 -0500 Subject: [PATCH] Checkpoint scoped live recovery owners and durable cohorts Co-Authored-By: Paperclip --- .../adapters/deadline-work.ts | 62 ++++++ .../active-run-watchdog/adapters/postgres.ts | 9 +- .../active-run-watchdog/application/ports.ts | 1 + .../active-run-watchdog/application/types.ts | 2 +- .../application/use-cases.ts | 12 +- .../src/modules/active-run-watchdog/index.ts | 13 +- server/src/services/activity-log.ts | 3 + .../services/delivery-work-notifications.ts | 8 + .../execution-control-notifications.ts | 7 +- server/src/services/heartbeat.ts | 4 + server/src/services/heartbeat/agent-status.ts | 4 +- .../services/heartbeat/live-recovery-work.ts | 119 +++++++++++ server/src/services/heartbeat/recovery.ts | 186 +++++++++++------- server/src/services/heartbeat/run-output.ts | 9 +- server/src/services/instance-settings.ts | 2 + server/src/services/issues.ts | 7 +- .../legacy-shutdown-workspace-settlement.ts | 4 +- .../live-run-recovery-notifications.ts | 25 +++ .../native-finalization-notifications.ts | 2 + .../src/services/native-local-process-stop.ts | 7 +- server/src/services/recovery/service.ts | 7 +- 21 files changed, 402 insertions(+), 91 deletions(-) create mode 100644 server/src/modules/active-run-watchdog/adapters/deadline-work.ts create mode 100644 server/src/services/heartbeat/live-recovery-work.ts create mode 100644 server/src/services/live-run-recovery-notifications.ts diff --git a/server/src/modules/active-run-watchdog/adapters/deadline-work.ts b/server/src/modules/active-run-watchdog/adapters/deadline-work.ts new file mode 100644 index 00000000000..5c264d25396 --- /dev/null +++ b/server/src/modules/active-run-watchdog/adapters/deadline-work.ts @@ -0,0 +1,62 @@ +import { and, asc, eq, gt, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.js"; +import type { LiveRunRecoveryScope } from "../../../services/live-run-recovery-notifications.js"; +import { registerScopedDeadlineWork } from "../../../services/scoped-deadline-work.js"; +import type { ScanSilentActiveRunsOptions } from "../application/use-cases.js"; +import type { WatchdogRunReader } from "../application/ports.js"; + +type Entry = { id: string; companyId: string; issueId: string | null; dueAt: number }; + +export function registerActiveRunSilenceWork(db: Db, + coordinator: ReturnType, + owner: { reader: WatchdogRunReader; scan(input: ScanSilentActiveRunsOptions): Promise; + suspicionThresholdMs: number; enabled(): Promise }, + canRun: () => boolean, +) { + const cursors = new Map(); + let enabled = false; + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.activeRunSilence, + reconstructionIntervalMs: 30_000, + beforePass: async () => { enabled = await owner.enabled(); }, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry, scope) => !scope || entry.companyId === scope.companyId && (!scope.runId || entry.id === scope.runId) && (!scope.issueId || entry.issueId === scope.issueId), + load: async scope => { + const key = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"; + if (!enabled) { cursors.delete(key); return []; } + const cursor = cursors.get(key); + const issueId = sql`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`; + const rows = await db.select({ issueId, id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + lastActivity: sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt}, + ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt})`.mapWith(heartbeatRuns.createdAt), + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "running"), + scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + scope?.issueId ? eq(issueId, scope.issueId) : undefined, + cursor ? gt(heartbeatRuns.id, cursor) : undefined, + )).orderBy(asc(heartbeatRuns.id)).limit(100); + const complete = rows.length < 100; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { complete, rows: rows.map(row => ({ id: row.id, companyId: row.companyId, issueId: row.issueId, + dueAt: row.lastActivity.getTime() + owner.suspicionThresholdMs })) }; + }, + check: async entry => { + if (!enabled) return null; + const run = await owner.reader.findRunForCompany(entry.companyId, entry.id); + if (!run || run.status !== "running") return null; + const started = run.lastOutputAt ?? run.processStartedAt ?? run.startedAt ?? run.createdAt; + if (!started) return null; + const due = started.getTime() + owner.suspicionThresholdMs; + if (due > Date.now()) return { ...entry, dueAt: due }; + const decision = await owner.reader.findLatestDecision(entry.companyId, entry.id, new Date()); + if (decision.quietUntilDecision) return { ...entry, dueAt: decision.quietUntilDecision.snoozedUntil.getTime() }; + let next: number | null = null; + await owner.scan({ companyId: entry.companyId, runId: entry.id, onDeferredDeadline: at => { + next = next === null ? at : Math.min(next, at); + } }); + return next === null ? null : { ...entry, dueAt: next }; + }, + }, canRun); +} diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 9531c449260..2fdc6f68f3c 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } from "../../../services/live-run-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { notifyExternalAgentSettlement } from "../../external-agents/index.js"; import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js"; @@ -100,6 +101,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch async function findCandidateSilentRuns(input: { companyId?: string; + runId?: string; suspicionBefore: Date; issueCreatedAtGte?: Date | null; }): Promise { @@ -109,6 +111,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch .where( and( input.companyId ? eq(heartbeatRuns.companyId, input.companyId) : undefined, + input.runId ? eq(heartbeatRuns.id, input.runId) : undefined, eq(heartbeatRuns.status, "running"), sql`coalesce(${heartbeatRuns.lastOutputAt}, ${heartbeatRuns.processStartedAt}, ${heartbeatRuns.startedAt}, ${heartbeatRuns.createdAt}) <= ${input.suspicionBefore.toISOString()}::timestamptz`, ), @@ -282,8 +285,9 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch } async function recordDecision(companyId: string, input: RecordDecisionInput): Promise { - const [row] = await db - .insert(heartbeatRunWatchdogDecisions) + const [row] = await db.transaction(async tx => { + await notifyActiveRunSilence(tx, { companyId, runId: input.runId }); + return tx.insert(heartbeatRunWatchdogDecisions) .values({ companyId, runId: input.runId, @@ -296,6 +300,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch createdByRunId: input.createdByRunId, }) .returning(); + }); await logActivity(db, { companyId, diff --git a/server/src/modules/active-run-watchdog/application/ports.ts b/server/src/modules/active-run-watchdog/application/ports.ts index 594f0285a1c..d574d196c02 100644 --- a/server/src/modules/active-run-watchdog/application/ports.ts +++ b/server/src/modules/active-run-watchdog/application/ports.ts @@ -13,6 +13,7 @@ import type { export type FindCandidateSilentRunsInput = { companyId?: string; + runId?: string; suspicionBefore: Date; issueCreatedAtGte?: Date | null; }; diff --git a/server/src/modules/active-run-watchdog/application/types.ts b/server/src/modules/active-run-watchdog/application/types.ts index 63fc3a8995e..b74eacb9b31 100644 --- a/server/src/modules/active-run-watchdog/application/types.ts +++ b/server/src/modules/active-run-watchdog/application/types.ts @@ -114,7 +114,7 @@ export type FoldSourceResolvedRunInput = { export type FoldOutcome = | { kind: "folded"; evaluationIssueId: string | null } - | { kind: "stale" }; + | { kind: "stale"; retryAt?: number }; export type WatchdogDecisionActor = | { type: "board"; userId?: string | null; runId?: string | null } diff --git a/server/src/modules/active-run-watchdog/application/use-cases.ts b/server/src/modules/active-run-watchdog/application/use-cases.ts index c8aa801ada3..9ea0bf24d88 100644 --- a/server/src/modules/active-run-watchdog/application/use-cases.ts +++ b/server/src/modules/active-run-watchdog/application/use-cases.ts @@ -100,7 +100,7 @@ export function createFoldSourceResolvedRun(deps: FoldSourceResolvedRunDeps) { fallbackProcessGroupId: input.run.processGroupId, }); - return deps.writer.foldSourceResolvedRun(input.run.companyId, { + const outcome = await deps.writer.foldSourceResolvedRun(input.run.companyId, { run: input.run, sourceIssue: input.sourceIssue, evidence: input.evidence, @@ -110,6 +110,8 @@ export function createFoldSourceResolvedRun(deps: FoldSourceResolvedRunDeps) { cleanup, now: input.now, }); + return outcome.kind === "stale" && cleanup.attempted && cleanup.outcome === "termination_sent_still_running" + ? { ...outcome, retryAt: input.now.getTime() + 30_000 } : outcome; }; } @@ -227,12 +229,14 @@ export type ScanSilentActiveRunsDeps = { export type ScanSilentActiveRunsOptions = { now?: Date; + runId?: string; + onDeferredDeadline?: (at: number) => void; companyId?: string; issueCreatedAtGte?: Date | null; }; type InspectOutcome = - | { kind: "skipped" } + | { kind: "skipped"; retryAt?: number } | { kind: "existing"; evaluationIssueId: string } | { kind: "folded"; evaluationIssueId: string | null }; @@ -283,7 +287,7 @@ export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) { }); return foldOutcome.kind === "folded" ? { kind: "folded", evaluationIssueId: foldOutcome.evaluationIssueId } - : { kind: "skipped" }; + : { kind: "skipped", ...(foldOutcome.retryAt === undefined ? {} : { retryAt: foldOutcome.retryAt }) }; } } @@ -305,6 +309,7 @@ export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) { const suspicionBefore = new Date(now.getTime() - deps.suspicionThresholdMs); const candidates = await deps.reader.findCandidateSilentRuns({ companyId: opts?.companyId, + runId: opts?.runId, suspicionBefore, issueCreatedAtGte: opts?.issueCreatedAtGte, }); @@ -331,6 +336,7 @@ export function createScanSilentActiveRuns(deps: ScanSilentActiveRunsDeps) { now, dismissedFalsePositive: decisionState.dismissedFalsePositive, }); + if (outcome.kind === "skipped" && outcome.retryAt !== undefined) opts?.onDeferredDeadline?.(outcome.retryAt); if (outcome.kind === "existing") result.existing += 1; else if (outcome.kind === "folded") result.folded += 1; else result.skipped += 1; diff --git a/server/src/modules/active-run-watchdog/index.ts b/server/src/modules/active-run-watchdog/index.ts index 907420badf9..92a1c9fe779 100644 --- a/server/src/modules/active-run-watchdog/index.ts +++ b/server/src/modules/active-run-watchdog/index.ts @@ -1,3 +1,4 @@ +import { registerActiveRunSilenceWork } from "./adapters/deadline-work.js"; import type { Db } from "@paperclipai/db"; import { createPostgresWatchdogAdapter } from "./adapters/postgres.js"; import { createProcessAdapter } from "./adapters/process.js"; @@ -29,17 +30,19 @@ export function createActiveRunWatchdog(db: Db, config: ActiveRunWatchdogConfig) processController, }); + const scanSilentActiveRuns = createScanSilentActiveRuns({ reader: postgresAdapter, foldSourceResolvedRun, + suspicionThresholdMs: config.suspicionThresholdMs }); + return { + registerSilenceWork: (coordinator: Parameters[1], canRun: () => boolean, + scan = scanSilentActiveRuns, enabled = async () => true) => registerActiveRunSilenceWork(db, coordinator, { reader: postgresAdapter, + scan, enabled, suspicionThresholdMs: config.suspicionThresholdMs }, canRun), buildRunOutputSilence: createBuildRunOutputSilence({ reader: postgresAdapter, suspicionThresholdMs: config.suspicionThresholdMs, criticalThresholdMs: config.criticalThresholdMs, }), - scanSilentActiveRuns: createScanSilentActiveRuns({ - reader: postgresAdapter, - foldSourceResolvedRun, - suspicionThresholdMs: config.suspicionThresholdMs, - }), + scanSilentActiveRuns, recordWatchdogDecision: createRecordWatchdogDecision({ reader: postgresAdapter, writer: postgresAdapter, diff --git a/server/src/services/activity-log.ts b/server/src/services/activity-log.ts index 5b10dac43b5..4e3aab320b0 100644 --- a/server/src/services/activity-log.ts +++ b/server/src/services/activity-log.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { beginRunPhase } from "./execution-preparation-timing.js"; import type { ActivityPublication } from "../types/activity-publication.js"; @@ -181,6 +182,8 @@ export async function persistActivity(db: Db, input: LogActivityInput) { const [activity] = input.entityType === "issue" ? await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, `task:${input.entityId}`); + if (input.runId && input.action === "issue.updated" && ["done", "cancelled"].includes(String(input.details?.status))) + await notifyActiveRunSilence(tx, { companyId: input.companyId, runId: input.runId }); return insert(tx as unknown as Db); }) : await insert(db); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 79f0f5d1d3d..8ca650525ad 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,14 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + nativeProcessSessions: "native-process-sessions", + nativeQuestionCancellations: "native-question-cancellations", + activeExecutionRecovery: "active-execution-recovery", + settledNativeProcesses: "settled-native-processes", + orphanActiveLeases: "orphan-active-leases", + legacyShutdownSettlement: "legacy-shutdown-settlement", + strandedAgentStatus: "stranded-agent-status", + activeRunSilence: "active-run-silence", deferredExecutionReceipts: "deferred-execution-receipts", taskRetryReceiptSettlement: "task-retry-receipt-settlement", nativeStatusWakeIntents: "native-status-wake-intents", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 7d9530f5197..615c6452327 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } 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"; @@ -9,6 +10,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut if (scope) { await notifyRunDispatch(transaction, scope); await notifyDeferredExecutionEligibility(transaction, scope); + await notifyActiveRunSilence(transaction, scope); } for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) { @@ -24,7 +26,10 @@ export function writeExecutionControl( ): Promise { return db.transaction(async transaction => { await notifyRunDispatch(transaction, scope); - if (scope) await notifyDeferredExecutionEligibility(transaction, scope); + if (scope) { + await notifyDeferredExecutionEligibility(transaction, scope); + await notifyActiveRunSilence(transaction, scope); + } await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors); for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic); return write(transaction as unknown as Db); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 63b836a566f..57328de4f7a 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4245,6 +4245,10 @@ export function heartbeatService( computeTaskDrain, applyTaskDrain, + registerActiveRunSilenceWork: (coordinator: Parameters[0], canRun: () => boolean) => + recovery.registerActiveRunSilenceWork(coordinator, () => canRun() && readTaskDrain(new Date()) === null, + async scope => recovery.scanSilentActiveRuns({ ...scope, issueCreatedAtGte: await getWorktreeExecutionCutoff() }), + async () => !(await getSchedulingSuppression(true)).suppressed), registerDeferredReceiptWork: (coordinator: Parameters[1], canRun: () => boolean) => registerDeferredReceiptWork(db, coordinator, { resume: scope => heartbeatQueue.resumeDeferredExecutionReceipts(scope), diff --git a/server/src/services/heartbeat/agent-status.ts b/server/src/services/heartbeat/agent-status.ts index f034cd06594..2957471926a 100644 --- a/server/src/services/heartbeat/agent-status.ts +++ b/server/src/services/heartbeat/agent-status.ts @@ -109,13 +109,15 @@ export function createHeartbeatAgentStatus(db: Db) { publishStatus(result.updated, outcome); } - async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet) { + async function reconcileStrandedAgentStatuses(liveRunIds: () => ReadonlySet, scope: { companyId?: string; agentId?: string } = {}) { const now = new Date(); const cutoff = new Date(now.getTime() - RECONCILIATION_GRACE_MS); const candidates = await db.transaction(async tx => { await tx.execute(sql`SET LOCAL statement_timeout = '2s'`); return tx.select({ id: agents.id, companyId: agents.companyId }).from(agents) .where(and(eq(agents.status, "running"), eq(agents.lifecycleState, "ready"), + scope.companyId ? eq(agents.companyId, scope.companyId) : undefined, + scope.agentId ? eq(agents.id, scope.agentId) : undefined, lte(agents.updatedAt, cutoff), settledAgentWorkCondition(liveRunIds(), now))) .orderBy(asc(agents.updatedAt), asc(agents.id)).limit(RECONCILIATION_BATCH_SIZE); }); diff --git a/server/src/services/heartbeat/live-recovery-work.ts b/server/src/services/heartbeat/live-recovery-work.ts new file mode 100644 index 00000000000..eb3e41204b1 --- /dev/null +++ b/server/src/services/heartbeat/live-recovery-work.ts @@ -0,0 +1,119 @@ +import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; +import { agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js"; +import type { LiveResourceRecoveryScope, LiveRunRecoveryScope } from "../live-run-recovery-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; + +const TERMINAL = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; +type Scope = LiveRunRecoveryScope & LiveResourceRecoveryScope; +type Entry = { id: string; companyId: string; runId?: string | null; dueAt: number }; +type CheckScope = { companyId: string; runId?: string; leaseId?: string; agentId?: string; + onDeferredDeadline(at: number): void; onError(error: unknown): void }; +type RunKind = "native" | "question" | "active" | "settled"; + +export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType, owner: { + native(scope: CheckScope): Promise; + question(scope: CheckScope): Promise; + active(scope: CheckScope): Promise; + settled(scope: CheckScope): Promise; + orphanLease(scope: CheckScope): Promise; + shutdown(scope: CheckScope): Promise; + agent(scope: CheckScope): Promise; + enabled(): Promise; +}, canRun: () => boolean) { + function register(queue: DeliveryQueue, kind: RunKind | "orphanLease" | "shutdown" | "agent") { + const cursors = new Map(); + 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 => { + const key = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, + leaseId: scope.leaseId, agentId: scope.agentId }) : "*"; + if (!enabled) { cursors.delete(key); return []; } + const cursor = cursors.get(key); + let rows: Entry[]; + if (kind === "agent") { + const candidates = await db.select({ id: agents.id, companyId: agents.companyId, updatedAt: agents.updatedAt }) + .from(agents).where(and(eq(agents.status, "running"), eq(agents.lifecycleState, "ready"), + scope ? eq(agents.companyId, scope.companyId) : undefined, + scope?.agentId ? eq(agents.id, scope.agentId) : undefined, + cursor ? gt(agents.id, cursor) : undefined, + )).orderBy(asc(agents.id)).limit(100); + rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, dueAt: row.updatedAt.getTime() + 5 * 60_000 })); + } else if (kind === "orphanLease" || kind === "shutdown") { + const candidates = await db.select({ id: environmentLeases.id, companyId: environmentLeases.companyId, + runId: environmentLeases.heartbeatRunId, updatedAt: environmentLeases.updatedAt, metadata: environmentLeases.metadata }).from(environmentLeases) + .leftJoin(heartbeatRuns, and(eq(heartbeatRuns.id, environmentLeases.heartbeatRunId), + eq(heartbeatRuns.companyId, environmentLeases.companyId))) + .where(and(scope ? eq(environmentLeases.companyId, scope.companyId) : undefined, + scope?.leaseId ? eq(environmentLeases.id, scope.leaseId) : undefined, + scope?.runId ? eq(environmentLeases.heartbeatRunId, scope.runId) : undefined, + cursor ? gt(environmentLeases.id, cursor) : undefined, + kind === "orphanLease" ? and(eq(environmentLeases.status, "active"), + or(isNull(environmentLeases.heartbeatRunId), inArray(heartbeatRuns.status, TERMINAL))) + : sql`${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,schema}' = 'paperclip.legacy-shutdown-workspace-settlement.v1' + and ${environmentLeases.metadata}#>>'{legacyShutdownWorkspaceSettlement,state}' = 'pending'`, + )).orderBy(asc(environmentLeases.id)).limit(100); + rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, runId: row.runId, + dueAt: kind === "orphanLease" ? row.updatedAt.getTime() + 5 * 60_000 + : Date.parse(String((row.metadata?.legacyShutdownWorkspaceSettlement as Record)?.deadlineAt)) || Date.now(), + })); + } else { + const candidates = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + nextAttemptAt: nativeRunFinalizations.nextAttemptAt, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }) + .from(heartbeatRuns).leftJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId))) + .where(and(scope ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope?.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, + cursor ? gt(heartbeatRuns.id, cursor) : undefined, + kind === "native" ? and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["running", "failed"]), + eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId)) + : kind === "question" ? and(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), + sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is not null`) + : kind === "active" ? eq(heartbeatRuns.status, "running") + : and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["failed", "cancelled", "timed_out", "interrupted"]), + isNotNull(heartbeatRuns.finishedAt), or(isNotNull(heartbeatRuns.processPid), + isNotNull(heartbeatRuns.processGroupId), isNotNull(heartbeatRuns.processStartedAt))), + )).orderBy(asc(heartbeatRuns.id)).limit(100); + rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, dueAt: kind === "native" + ? Math.max(Date.now(), row.nextAttemptAt?.getTime() ?? 0, row.leaseExpiresAt?.getTime() ?? 0) : Date.now() })); + } + const complete = rows.length < 100; + if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id); + return { complete, rows }; + }, + check: async entry => { + if (!enabled) return null; + let next: number | null = null; + const errors: unknown[] = []; + const scope: CheckScope = { companyId: entry.companyId, + ...(kind === "agent" ? { agentId: entry.id } : kind === "orphanLease" || kind === "shutdown" + ? { leaseId: entry.id } : { runId: entry.id }), + onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); }, + onError: error => { errors.push(error); }, + }; + await owner[kind](scope); + if (errors.length) throw new AggregateError(errors, `Recovery obligation ${entry.id} failed`); + return next === null ? null : { ...entry, dueAt: next }; + }, + }, canRun); + } + const workers = [register(DELIVERY_QUEUES.nativeProcessSessions, "native"), + register(DELIVERY_QUEUES.nativeQuestionCancellations, "question"), + register(DELIVERY_QUEUES.activeExecutionRecovery, "active"), + register(DELIVERY_QUEUES.settledNativeProcesses, "settled"), + register(DELIVERY_QUEUES.orphanActiveLeases, "orphanLease"), + register(DELIVERY_QUEUES.legacyShutdownSettlement, "shutdown"), + register(DELIVERY_QUEUES.strandedAgentStatus, "agent")]; + return { ready: Promise.all(workers.map(worker => worker.ready)), + wake() { for (const worker of workers) worker.invalidate(); }, + async stop() { await Promise.all(workers.map(worker => worker.stop())); } }; +} diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index e506f0f7e62..141d0f17755 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -22,7 +22,7 @@ import { discoverSettledNativeRunProcesses, recordNativeLocalProcessStop, retryP import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; import { legacyControllerBootId, - hasLiveLegacyController, + liveLegacyControllerExpiresAt, revokeExpiredLegacyControllerWithState, } from "../legacy-controller-lease.js"; import { remoteTerminationReceipt } from "../remote-execution-termination.js"; @@ -1493,6 +1493,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // data already on the lease row. async function sweepOrphanedActiveLeases(opts: { backoffMs: number; + companyId?: string; + leaseId?: string; }): Promise<{ recovered: number }> { const cutoff = new Date(Date.now() - opts.backoffMs); @@ -1506,6 +1508,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD .where( and( eq(environmentLeases.status, "active"), + opts.companyId ? eq(environmentLeases.companyId, opts.companyId) : undefined, + opts.leaseId ? eq(environmentLeases.id, opts.leaseId) : undefined, or( isNull(environmentLeases.heartbeatRunId), inArray(heartbeatRuns.status, [ @@ -1977,59 +1981,15 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD void cleanup.finally(() => activeRunExecutionPromises.delete(cleanup)); } - let settledProcessesDiscovered = false; + type RecoveryScope = { + companyId?: string; + runId?: string; + onDeferredDeadline?: (at: number) => void; + onError?: (error: unknown) => void; + }; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipNativeStatusWakeIntents?: boolean; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) { - if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); - const staleThresholdMs = opts?.staleThresholdMs ?? 0; + async function resumeRetryableNativeProcessSessions(scope: RecoveryScope = {}, beforeDispatch?: () => Promise) { const now = new Date(); - if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep(); - // Recovery never launches a provider or infers stopped ownership from - // terminal status. Uncaptured local copies require durable stop evidence. - if (!opts?.skipStoppedInstructionCollection) await instructionCopies.recoverStopped().catch(error => { - logger.warn({ err: error }, "failed to recover stopped instruction copies"); - }); - await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit, skipDirectoryCleanup: opts?.skipInstructionDirectoryCleanup }).catch(error => { - logger.warn({ err: error }, "failed to retry captured instruction revisions"); - }); - - // Complete persisted native results before generic orphan recovery. The - // reconciler reads the durable workspace barrier and persisted runtime - // mode, never the current feature flag. - await reconcileNativeFinalizations(db, undefined, { - skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews, - skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews, - skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, - skipNativePendingEffects: opts?.skipNativePendingEffects, - skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, - skipNativeReviewResponse: opts?.skipNativeReviewResponse, - skipNativeStatusReassessment: opts?.skipNativeStatusReassessment, - skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization, - skipNativeResultCommit: opts?.skipNativeResultCommit, - skipNativeRunProjection: opts?.skipNativeRunProjection, - skipExternalConversationSettlement: opts?.skipExternalConversationSettlement, - skipExternalAgentSettlement: opts?.skipExternalAgentSettlement, - skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices, - environmentRuntime, - pluginWorkerManager: dependencies.pluginWorkerManager, - onWorkspaceSettled: input => settleRecoveredNativeWorkspace(input, { - skipEnvironmentSettlement: opts?.skipNativeEnvironmentSettlement, - }), - }).catch((error) => { - logger.warn( - { err: error }, - "failed to reconcile persisted native finalizations before orphan reaping", - ); - }); - if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup(); - await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); - if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => { - logger.warn( - { err: error }, - "failed to dispatch persisted native status wake intents before orphan reaping", - ); - }); - // A retryable native run can retain process identifiers from the failed // attempt. Inspect them before the recovery claim: a live identifier is // unowned and blocks recovery, while identifiers that are all dead can be @@ -2044,6 +2004,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ) .where( and( + scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, inArray(heartbeatRuns.status, ["running", "failed"]), eq(heartbeatRuns.runtimeMode, "native"), eq(nativeRunFinalizations.phase, "retryable_failure"), @@ -2067,6 +2029,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD processPidAlive, processGroupAlive, }); + scope.onDeferredDeadline?.(Date.now() + 30_000); continue; } const cleared = await db.transaction(async tx => { @@ -2101,17 +2064,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD if (cleared) claimableNativeRunIds.add(cleared.id); } - const executingRun = (runId: string) => activeRunExecutions.has(runId) || runningProcesses.has(runId); - if (!settledProcessesDiscovered) { - settledProcessesDiscovered = await discoverSettledNativeRunProcesses(db, { skip: executingRun }).then(() => true, (error) => { - logger.warn({ err: error }, "failed to discover settled native process identities"); - return false; - }); - } else { - await retryPendingSettledNativeRunProcesses(db, { skip: executingRun }).catch((error) => { - logger.warn({ err: error }, "failed to retry settled native process identity release"); - }); - } + await beforeDispatch?.(); // An explicit result-less retryable failure resumes on the original run. // The database lease is claimed before dispatch so concurrent service @@ -2141,6 +2094,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ); }, }).catch((error) => { + scope.onError?.(error); logger.warn( { err: error }, "failed to claim persisted native session resumptions", @@ -2151,6 +2105,10 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD nativeResumeClaims.map((claim) => claim.runId), ); + return resumedRunIds; + } + + async function recoverQuestionCancellationRequests(scope: RecoveryScope = {}) { // A terminal issue transition writes this intent in the same transaction // that closes the question's task, even when its card is retained. Consume // it before generic orphan recovery so a restart preserves cancellation. @@ -2162,6 +2120,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD .from(heartbeatRuns) .where( and( + scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined, inArray(heartbeatRuns.status, [ ...CANCELLABLE_HEARTBEAT_RUN_STATUSES, ]), @@ -2200,6 +2160,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }, }); } catch (err) { + scope.onError?.(err); // Keep the marker intact for the next startup/periodic sweep. logger.warn( { err, runId: request.id }, @@ -2208,6 +2169,14 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } } + } + + async function recoverActiveExecutions(scope: RecoveryScope = {}, options: { + staleThresholdMs?: number; resumedRunIds?: Set; + } = {}) { + const now = new Date(); + const staleThresholdMs = options.staleThresholdMs ?? 5 * 60_000; + const resumedRunIds = options.resumedRunIds ?? new Set(); // Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them) const activeRuns = await db .select({ @@ -2229,7 +2198,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id), ) - .where(eq(heartbeatRuns.status, "running")); + .where(and(eq(heartbeatRuns.status, "running"), + scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined, + scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined)); // Reuse the orphan sweep's active-run inventory. Only an expired, untracked // controller creates recovery work; idle sweeps add no checkpoint query. @@ -2362,12 +2333,16 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } if (resumedRunIds.has(run.id)) continue; if (locallyTracked) continue; - if (await hasLiveLegacyController(db, run)) continue; + const legacyLease = await liveLegacyControllerExpiresAt(db, run); + if (legacyLease) { scope.onDeferredDeadline?.(legacyLease.getTime()); continue; } // Apply staleness threshold to avoid false positives if (staleThresholdMs > 0) { const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0; - if (now.getTime() - refTime < staleThresholdMs) continue; + if (now.getTime() - refTime < staleThresholdMs) { + scope.onDeferredDeadline?.(refTime + staleThresholdMs); + continue; + } } const currentAdapterTracksLocalChild = @@ -2392,6 +2367,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD (processPidAlive || processGroupAlive) && readHotRestartAdoptionMetadata(parseObject(run.resultJson)) ) { + scope.onDeferredDeadline?.(Date.now() + 30_000); continue; } if (processPidAlive || processGroupAlive) { @@ -2417,6 +2393,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }); } } + scope.onDeferredDeadline?.(Date.now() + 30_000); continue; } @@ -2566,6 +2543,80 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ); } + return reaped; + } + + let settledProcessesDiscovered = false; + + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipNativeStatusWakeIntents?: boolean; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean; skipNativeStatusReassessment?: boolean; skipRetainedNativeSessionCleanup?: boolean; skipNativeWorkspaceFinalization?: boolean; skipNativeResultCommit?: boolean; skipNativeRunProjection?: boolean; skipNativeRecoveryNotices?: boolean; skipExternalConversationSettlement?: boolean; skipExternalAgentSettlement?: boolean; skipNativeEnvironmentSettlement?: boolean }) { + if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); + const staleThresholdMs = opts?.staleThresholdMs ?? 0; + const now = new Date(); + if (!opts?.skipScheduledRetryCancellation) await scheduledRetryCancellation.sweep(); + // Recovery never launches a provider or infers stopped ownership from + // terminal status. Uncaptured local copies require durable stop evidence. + if (!opts?.skipStoppedInstructionCollection) await instructionCopies.recoverStopped().catch(error => { + logger.warn({ err: error }, "failed to recover stopped instruction copies"); + }); + await instructionCopies.recoverCaptured({ skipCommit: opts?.skipCapturedInstructionCommit, skipDirectoryCleanup: opts?.skipInstructionDirectoryCleanup }).catch(error => { + logger.warn({ err: error }, "failed to retry captured instruction revisions"); + }); + + // Complete persisted native results before generic orphan recovery. The + // reconciler reads the durable workspace barrier and persisted runtime + // mode, never the current feature flag. + await reconcileNativeFinalizations(db, undefined, { + skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews, + skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews, + skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, + skipNativePendingEffects: opts?.skipNativePendingEffects, + skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, + skipNativeReviewResponse: opts?.skipNativeReviewResponse, + skipNativeStatusReassessment: opts?.skipNativeStatusReassessment, + skipNativeWorkspaceFinalization: opts?.skipNativeWorkspaceFinalization, + skipNativeResultCommit: opts?.skipNativeResultCommit, + skipNativeRunProjection: opts?.skipNativeRunProjection, + skipExternalConversationSettlement: opts?.skipExternalConversationSettlement, + skipExternalAgentSettlement: opts?.skipExternalAgentSettlement, + skipNativeRecoveryNotices: opts?.skipNativeRecoveryNotices, + environmentRuntime, + pluginWorkerManager: dependencies.pluginWorkerManager, + onWorkspaceSettled: input => settleRecoveredNativeWorkspace(input, { + skipEnvironmentSettlement: opts?.skipNativeEnvironmentSettlement, + }), + }).catch((error) => { + logger.warn( + { err: error }, + "failed to reconcile persisted native finalizations before orphan reaping", + ); + }); + if (!opts?.skipRetainedNativeSessionCleanup) scheduleRetainedNativeSessionCleanup(); + await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); + if (!opts?.skipNativeStatusWakeIntents) await dispatchPendingNativeStatusWakeups().catch((error) => { + logger.warn( + { err: error }, + "failed to dispatch persisted native status wake intents before orphan reaping", + ); + }); + + const resumedRunIds = await resumeRetryableNativeProcessSessions({}, async () => { + const executingRun = (runId: string) => activeRunExecutions.has(runId) || runningProcesses.has(runId); + if (!settledProcessesDiscovered) { + settledProcessesDiscovered = await discoverSettledNativeRunProcesses(db, { skip: executingRun }).then(() => true, (error) => { + logger.warn({ err: error }, "failed to discover settled native process identities"); + return false; + }); + } else { + await retryPendingSettledNativeRunProcesses(db, { skip: executingRun }).catch((error) => { + logger.warn({ err: error }, "failed to retry settled native process identity release"); + }); + } + + }); + + await recoverQuestionCancellationRequests(); + const reaped = await recoverActiveExecutions({}, { staleThresholdMs, resumedRunIds }); + // Recover an active lease whose run already ended before the same-tick // pending_cleanup sweep, so this tick can stop the recovered sandbox. // Isolate the sweep so its failure never hides the reaper result. @@ -2634,6 +2685,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD reconcileHotRestartAdoption, recoverNativeRunsAfterRestart, reapOrphanedRuns, + resumeRetryableNativeProcessSessions, + recoverQuestionCancellationRequests, + recoverActiveExecutions, sweepOrphanedActiveLeases, drainRunningRunsForShutdown, }; diff --git a/server/src/services/heartbeat/run-output.ts b/server/src/services/heartbeat/run-output.ts index 71a606a53a4..24a607297bf 100644 --- a/server/src/services/heartbeat/run-output.ts +++ b/server/src/services/heartbeat/run-output.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } from "../live-run-recovery-notifications.js"; import { eq } from "drizzle-orm"; import { heartbeatRuns, type Db } from "@paperclipai/db"; import type { AdapterInvocationMeta, AdapterRuntimeEvent } from "../../adapters/index.js"; @@ -94,8 +95,9 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput) pendingOutputProgress.at.getTime() - lastOutputFlushAt.getTime() >= ACTIVE_RUN_OUTPUT_PROGRESS_FLUSH_INTERVAL_MS; if (!shouldFlush) return; - progressWrite = db - .update(heartbeatRuns) + progressWrite = db.transaction(async tx => { + await notifyActiveRunSilence(tx, { companyId: run.companyId, runId: run.id }); + await tx.update(heartbeatRuns) .set({ lastOutputAt: pendingOutputProgress.at, lastOutputSeq: pendingOutputProgress.seq, @@ -103,7 +105,8 @@ export function createHeartbeatRunOutput(db: Db, input: HeartbeatRunOutputInput) lastOutputBytes: pendingOutputProgress.bytes, updatedAt: new Date(), }) - .where(eq(heartbeatRuns.id, run.id)) + .where(eq(heartbeatRuns.id, run.id)); + }) .then(() => { lastOutputFlushAt = pendingOutputProgress.at; if (outputProgressState.pending === pendingOutputProgress) diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 6c5f948d303..a3e2b2f3b14 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js"; @@ -634,6 +635,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService if (patch.enableWorktreeRunExecution !== undefined) { await notifyRunDispatch(tx); await notifyDeferredExecutionEligibility(tx); + await notifyActiveRunSilence(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyExternalAgentSettlement(tx); await notifyInstructionCommit(tx); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index fa47af138b9..8955ace27c8 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js"; import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js"; import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; @@ -11507,6 +11508,7 @@ export function issueService(db: Db) { || blockedByIssueIds !== undefined) { await notifyRunDispatch(tx, { companyId: receiptExisting.companyId }); await notifyDeferredExecutionEligibility(tx, { companyId: receiptExisting.companyId }); + await notifyActiveRunSilence(tx, { companyId: receiptExisting.companyId, issueId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); } if (patch.monitorNextCheckAt !== undefined || (receiptExisting.monitorNextCheckAt && @@ -11972,7 +11974,10 @@ export function issueService(db: Db) { await notifyDecisionTargetIssueRemoval(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); await notifyRunDispatch(tx, owner ? { companyId: owner.companyId } : undefined); - if (owner) await notifyDeferredExecutionEligibility(tx, owner); + if (owner) { + await notifyDeferredExecutionEligibility(tx, owner); + await notifyActiveRunSilence(tx, { companyId: owner.companyId, issueId: id }); + } await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); removedIssue = await tx .delete(issues) diff --git a/server/src/services/legacy-shutdown-workspace-settlement.ts b/server/src/services/legacy-shutdown-workspace-settlement.ts index 54f9b1a1c00..8c85cd3f14b 100644 --- a/server/src/services/legacy-shutdown-workspace-settlement.ts +++ b/server/src/services/legacy-shutdown-workspace-settlement.ts @@ -73,8 +73,10 @@ export async function legacyShutdownWorkspaceResourceProtected(db: Db, } /** Released stop receipts still need reconciliation if export never settled. */ -export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date()) { +export async function sweepLegacyShutdownWorkspaceSettlements(db: Db, now = new Date(), scope: { companyId?: string; leaseId?: string } = {}) { const rows = await db.select().from(environmentLeases).where(and( + scope.companyId ? eq(environmentLeases.companyId, scope.companyId) : undefined, + scope.leaseId ? eq(environmentLeases.id, scope.leaseId) : undefined, sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'schema' = ${SCHEMA}`, sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'state' = 'pending'`, sql`${environmentLeases.metadata}->'legacyShutdownWorkspaceSettlement'->>'deadlineAt' <= ${now.toISOString()}`, diff --git a/server/src/services/live-run-recovery-notifications.ts b/server/src/services/live-run-recovery-notifications.ts new file mode 100644 index 00000000000..10334f6776c --- /dev/null +++ b/server/src/services/live-run-recovery-notifications.ts @@ -0,0 +1,25 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type LiveRunRecoveryScope = { companyId: string; runId?: string; issueId?: string; agentId?: string }; + +export async function notifyActiveRunSilence(tx: object, scope?: LiveRunRecoveryScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.activeRunSilence, scope + ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, issueId: scope.issueId }) : "*"); +} + +export async function notifyLiveRunRecovery(tx: object, scope?: LiveRunRecoveryScope) { + const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, agentId: scope.agentId }) : "*"; + for (const queue of [DELIVERY_QUEUES.nativeProcessSessions, DELIVERY_QUEUES.nativeQuestionCancellations, + DELIVERY_QUEUES.activeExecutionRecovery, DELIVERY_QUEUES.settledNativeProcesses, DELIVERY_QUEUES.strandedAgentStatus]) + await notifyDeliveryWork(tx, queue, hint); + await notifyActiveRunSilence(tx, scope); +} + +export type LiveResourceRecoveryScope = { companyId: string; runId?: string; leaseId?: string }; + +export async function notifyLiveResourceRecovery(tx: object, scope?: LiveResourceRecoveryScope) { + const hint = scope ? JSON.stringify({ companyId: scope.companyId, runId: scope.runId, leaseId: scope.leaseId }) : "*"; + for (const queue of [DELIVERY_QUEUES.orphanActiveLeases, DELIVERY_QUEUES.legacyShutdownSettlement, + DELIVERY_QUEUES.strandedAgentStatus, DELIVERY_QUEUES.settledNativeProcesses]) + await notifyDeliveryWork(tx, queue, hint); +} diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index 72d4cbcac49..a38aae46c0b 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,3 +1,4 @@ +import { notifyActiveRunSilence } 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"; @@ -15,6 +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 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 0d31853d552..b26a29e7add 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -145,7 +145,7 @@ export async function readNativeLocalProcessStop(db: Db, companyId: string, runI type SettledProcessRelease = "released" | "pending" | "skipped"; -export async function releaseSettledNativeRunProcess(db: Db, runId: string): Promise { +export async function releaseSettledNativeRunProcess(db: Db, runId: string, options: { onDeferredDeadline?: (at: number) => void } = {}): Promise { return db.transaction(async transaction => { const tx = transaction as unknown as Db; const [run] = await tx.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, @@ -173,7 +173,10 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string): Pro )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); if (identity?.payload?.targetKind !== "local" || identity.payload.hostname !== os.hostname() || identity.payload.processPid !== run.processPid || identity.payload.processGroupId !== run.processGroupId) return "skipped"; - if (!(await recordNativeLocalProcessStop(tx, run))) return "pending"; + if (!(await recordNativeLocalProcessStop(tx, run))) { + options.onDeferredDeadline?.(Date.now() + 30_000); + return "pending"; + } } await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id }); await tx.update(heartbeatRuns).set({ diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 4b2ec5c0672..df9b7f84333 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -2384,11 +2384,7 @@ export function recoveryService( }); } - async function scanSilentActiveRuns(opts?: { - now?: Date; - companyId?: string; - issueCreatedAtGte?: Date | null; - }) { + async function scanSilentActiveRuns(opts?: Parameters[0]) { return watchdog.scanSilentActiveRuns(opts); } @@ -6568,6 +6564,7 @@ export function recoveryService( escalateStrandedAssignedIssue, recordWatchdogDecision, scanSilentActiveRuns, + registerActiveRunSilenceWork: watchdog.registerSilenceWork, reconcileStrandedAssignedIssues, reconcileUnassignedBlockingIssues, reconcileActiveRecoveryActions,