Drive remaining live run and resource recovery from durable work (#18419)

Reviewed and verified at 6a4c193906.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-31 16:00:35 -05:00
1 parent eb12cfc96b
commit 9ca9af544a
62 files changed
+1822 -237

No files matched your search

@@ -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<typeof createDeliveryWorkCoordinator>,
owner: { reader: WatchdogRunReader; scan(input: ScanSilentActiveRunsOptions): Promise<unknown>;
suspicionThresholdMs: number; enabled(): Promise<boolean> },
canRun: () => boolean,
) {
const cursors = new Map<string, string>();
let enabled = false;
return registerScopedDeadlineWork<Entry, LiveRunRecoveryScope>(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<string | null>`coalesce(${heartbeatRuns.contextSnapshot}->>'issueId', ${heartbeatRuns.contextSnapshot}->>'taskId')`;
const rows = await db.select({ issueId, id: heartbeatRuns.id, companyId: heartbeatRuns.companyId,
lastActivity: sql<Date>`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);
}
@@ -1,3 +1,4 @@
import { notifyActiveRunSilence, notifyLiveRunRecovery, notifyLiveResourceRecovery } from "../../../services/live-run-recovery-notifications.js";
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
@@ -101,6 +102,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
async function findCandidateSilentRuns(input: {
companyId?: string;
runId?: string;
suspicionBefore: Date;
issueCreatedAtGte?: Date | null;
}): Promise<RunSnapshot[]> {
@@ -110,6 +112,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`,
),
@@ -283,8 +286,9 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
}
async function recordDecision(companyId: string, input: RecordDecisionInput): Promise<WatchdogDecisionRecord> {
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,
@@ -297,6 +301,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
createdByRunId: input.createdByRunId,
})
.returning();
});
await logActivity(db, {
companyId,
@@ -344,6 +349,8 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
};
const transactionResult = await db.transaction(async (tx) => {
await notifyLiveRunRecovery(tx, input.run);
await notifyLiveResourceRecovery(tx, input.run);
await notifyInstructionCollection(tx, { companyId, runId: input.run.id });
await notifyExternalAgentSettlement(tx, { companyId, runId: input.run.id });
await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id });
@@ -13,6 +13,7 @@ import type {
export type FindCandidateSilentRunsInput = {
companyId?: string;
runId?: string;
suspicionBefore: Date;
issueCreatedAtGte?: Date | null;
};
@@ -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 }
@@ -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;
@@ -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<typeof registerActiveRunSilenceWork>[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,
@@ -1,3 +1,4 @@
import { notifyExecutionRunWork } from "../../../services/execution-control-notifications.js";
import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js";
import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js";
import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js";
@@ -687,6 +688,7 @@ export function createPostgresRunDispatchAdapter(
| { applied: true; run: HeartbeatRun; postCommitEffects: PostCommitEffect[] }
| { applied: false }
> {
await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId });
await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId });
await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId });
@@ -732,6 +734,7 @@ export function createPostgresRunDispatchAdapter(
tx: Db,
input: CancelSuppressedRetryInput,
): Promise<CancelSuppressedRetryResult> {
await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId });
await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
await notifyExternalAgentSettlement(tx, { companyId: input.companyId, runId: input.runId });
await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId });
@@ -1018,6 +1021,7 @@ export function createPostgresRunDispatchAdapter(
now: Date,
providerDispatchNotStarted = false,
): Promise<CancelStaleQueuedRunOutcome> {
await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id });
await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
@@ -871,7 +871,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
expect(candidate?.id).toBe(wakeId);
const agent = await ports.transaction.findInvokableAgent({ companyId, agentId: foreignAgentId });
expect(agent).toBeNull();
const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, now: new Date() });
const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, agentId: candidate!.agentId, now: new Date() });
expect(failed).toBe(true);
return { outcome: { kind: "released" as const }, postCommitEffects: [] };
});
@@ -903,6 +903,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
async (_locked, ports) => {
const cancelledUnderWrongCompany = await ports.transaction.cancelDeferredWake({
companyId: otherCompanyId,
agentId,
wakeId,
reason: "cross-company cancel attempt",
now: new Date(),
@@ -911,6 +912,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
const failedUnderWrongCompany = await ports.transaction.failDeferredWake({
companyId: otherCompanyId,
agentId,
wakeId,
now: new Date(),
});
@@ -1,3 +1,4 @@
import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js";
import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js";
import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js";
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
@@ -275,6 +276,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
throw error;
}
const now = new Date();
await notifyStrandedAgentStatus(tx, { companyId, agentId: row.agentId });
await notifyChatInboundWakeNotices(tx, { companyId, issueId });
await tx
.update(agentWakeupRequests)
@@ -320,7 +322,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
return { liveNonSelfCommentIds, containedSelfAuthoredComment };
},
async cancelDeferredWake({ companyId, wakeId, reason, now }) {
async cancelDeferredWake({ companyId, wakeId, agentId, reason, now }) {
await notifyStrandedAgentStatus(tx, { companyId, agentId });
await notifyChatInboundWakeNotices(tx, { companyId });
const rows = await tx
.update(agentWakeupRequests)
@@ -328,6 +331,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
.where(
and(
eq(agentWakeupRequests.id, wakeId),
eq(agentWakeupRequests.agentId, agentId),
eq(agentWakeupRequests.companyId, companyId),
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
),
@@ -352,7 +356,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
return row ? toDeferredWakeCandidate(row) : null;
},
async failDeferredWake({ companyId, wakeId, now }) {
async failDeferredWake({ companyId, wakeId, agentId, now }) {
await notifyStrandedAgentStatus(tx, { companyId, agentId });
await notifyChatInboundWakeNotices(tx, { companyId });
const rows = await tx
.update(agentWakeupRequests)
@@ -365,6 +370,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
.where(
and(
eq(agentWakeupRequests.id, wakeId),
eq(agentWakeupRequests.agentId, agentId),
eq(agentWakeupRequests.companyId, companyId),
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
),
@@ -1,3 +1,5 @@
import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js";
import { notifyExecutionRunWork } from "../../../services/execution-control-notifications.js";
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
@@ -53,7 +55,7 @@ export type QueuedCommentQueuePostgresAdapterDeps = {
syncCommentExternalObjectsSafely(commentId: string, tx: Db): Promise<void>;
};
function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction {
function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps, wakeAgentId: string): QueuedCommentQueueTransaction {
return {
async updateCommentBody({ issueId, commentId, body, updatedAt }) {
await notifyStatusCardIssueChange(tx, companyId);
@@ -103,6 +105,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
},
async cancelWake({ wakeId, reason, now }) {
await notifyStrandedAgentStatus(tx, { companyId, agentId: wakeAgentId });
await notifyChatInboundWakeNotices(tx, { companyId, issueId });
await tx
.update(agentWakeupRequests)
@@ -111,6 +114,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
},
async cancelQueueRun({ queueRunId, reason, now }) {
await notifyExecutionRunWork(tx, { companyId, runId: queueRunId });
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId });
await notifyInstructionCollection(tx, { companyId, runId: queueRunId });
await notifyExternalAgentSettlement(tx, { companyId, runId: queueRunId });
@@ -311,7 +315,7 @@ export function createQueuedCommentIssueLockWriter(db: Db, deps: QueuedCommentQu
.then((rows) => rows[0] ?? null)
: null;
const transaction = buildTransaction(tx, companyId, input.issue.id, deps);
const transaction = buildTransaction(tx, companyId, input.issue.id, deps, wakeRow.agentId);
const wake = toWakeRow(wakeRow);
const queueRun = queueRunRow ? toRunRow(queueRunRow) : null;
const activeRun = activeRunRow ? toRunRow(activeRunRow) : null;
@@ -123,6 +123,7 @@ export interface WakeQueueTransaction {
cancelDeferredWake(input: {
companyId: string;
wakeId: string;
agentId: string;
reason: string;
now: Date;
}): Promise<boolean>;
@@ -135,7 +136,7 @@ export interface WakeQueueTransaction {
now: Date;
}): Promise<DeferredWakeCandidate | null>;
/** Sets `status = 'failed'` guarded by the current `deferred_issue_execution` status. */
failDeferredWake(input: { companyId: string; wakeId: string; now: Date }): Promise<boolean>;
failDeferredWake(input: { companyId: string; wakeId: string; agentId: string; now: Date }): Promise<boolean>;
getPauseHoldFacts(input: {
companyId: string;
issueId: string;
@@ -191,6 +191,7 @@ async function runReleaseDrain(
await ports.transaction.cancelDeferredWake({
companyId: run.companyId,
wakeId: candidate.id,
agentId: candidate.agentId,
reason: "Delegation closing note refers to completed child work",
now: input.now,
});
@@ -212,6 +213,7 @@ async function runReleaseDrain(
await ports.transaction.cancelDeferredWake({
companyId: run.companyId,
wakeId: candidate.id,
agentId: candidate.agentId,
reason: "Deferred task messages now belong to the current assignee",
now: input.now,
});
@@ -260,6 +262,7 @@ async function runReleaseDrain(
await ports.transaction.cancelDeferredWake({
companyId: run.companyId,
wakeId: candidate.id,
agentId: candidate.agentId,
reason: commentAction.selfAuthored
? "Deferred wake contained only comments authored by the finishing run"
: "Queued messages were discarded before promotion",
@@ -288,7 +291,7 @@ async function runReleaseDrain(
});
if (wakeOutcome.kind === "fail_not_invokable") {
await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, now: input.now });
await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, agentId: workingCandidate.agentId, now: input.now });
continue;
}
@@ -300,7 +303,7 @@ async function runReleaseDrain(
if (wakeOutcome.kind === "cancel_pause_hold") {
await ports.transaction.cancelDeferredWake({
companyId: run.companyId,
wakeId: workingCandidate.id,
wakeId: workingCandidate.id, agentId: workingCandidate.agentId,
reason: "Deferred wake suppressed by active subtree pause hold",
now: input.now,
});
@@ -374,7 +377,7 @@ async function promoteDeferredWake(
) {
await ports.transaction.cancelDeferredWake({
companyId: run.companyId,
wakeId: workingCandidate.id,
wakeId: workingCandidate.id, agentId: workingCandidate.agentId,
reason: "Deferred execution wake no longer applies to a terminal task",
now: input.now,
});