From dc4456c76b3eaf261be7568c45940769b23c0236 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 18:10:00 -0500 Subject: [PATCH] Drive issue recovery from scoped source writes and saved deadlines --- doc/plans/2026-10-30-issue-recovery-sleep.md | 18 ++ .../issue-recovery-deadlines.test.ts | 237 ++++++++++++++++++ .../issue-recovery-work-lifecycle.test.ts | 99 ++++++++ .../server-startup-feedback-export.test.ts | 6 +- server/src/index.ts | 76 +----- server/src/services/agent-conversations.ts | 2 + .../telegram/attachment-failure.ts | 2 + server/src/services/company-deletion.ts | 2 + server/src/services/company-portability.ts | 6 +- server/src/services/documents.ts | 11 + ...l-conversation-settlement-notifications.ts | 4 +- server/src/services/issue-recovery-actions.ts | 6 + server/src/services/issue-tree-control.ts | 8 +- server/src/services/issues.ts | 3 + .../native-runtime/chat-attachment-reuse.ts | 6 +- .../native-runner-file-handoff.ts | 6 +- ...ative-status-reassessment-notifications.ts | 4 +- server/src/services/system-recovery.ts | 2 + server/src/services/task-watchdog-work.ts | 2 +- server/src/services/task-watchdogs.ts | 6 +- 20 files changed, 419 insertions(+), 87 deletions(-) create mode 100644 doc/plans/2026-10-30-issue-recovery-sleep.md create mode 100644 server/src/__tests__/issue-recovery-deadlines.test.ts create mode 100644 server/src/__tests__/issue-recovery-work-lifecycle.test.ts diff --git a/doc/plans/2026-10-30-issue-recovery-sleep.md b/doc/plans/2026-10-30-issue-recovery-sleep.md new file mode 100644 index 00000000000..18670cc66b8 --- /dev/null +++ b/doc/plans/2026-10-30-issue-recovery-sleep.md @@ -0,0 +1,18 @@ +# Issue recovery without periodic scans + +Four existing owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. + +| Owner | Durable truth and producer boundary | Deadline or event | Exit | +| --- | --- | --- | --- | +| Stranded assignment | Visible assigned todo/in-progress/review issues; issue create/update/checkout/release; run lifecycle and execution control; agent/company/budget/configuration; interaction, approval and subtree hold changes | A committed eligibility change. A transient continuation uses the latest finish plus the existing exponential delay. | An execution path, durable wait, escalation, or current ineligibility. | +| Resolved dependency | Blocked assigned issues and company-scoped blocker relations; blocker status or relation changes through issue service and import; run/control and pause gate changes | The committing change. A failed enqueue retains that issue with error backoff. | Existing idempotent wake, live path, unresolved dependency or deliberate self-block. | +| Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. | +| Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. | + +The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry. + +Producer coverage uses the existing run-dispatch, execution-control, issue-review-state, native status reassessment and external conversation settlement boundaries. Direct gaps have explicit signals inside their owning transaction: document writes/import, attachment removal, blocker import, recovery-action adoption/mutation, conversation pause release, Telegram failure comment replacement, native artifact title/metadata updates and company deletion. Metadata-only comment classification that does not change timestamp, visibility or execution eligibility needs no additional watchdog activation. Approval links and decisions use their existing issue-review/continuation effects. Work-product create/update/delete/import uses its existing native status boundary. + +The new owners replace only these four startup and periodic calls. Deferred comments, silent active runs and residual orphan/resource recovery retain their existing backstop until the next slice. Every previously migrated native-owner skip remains unchanged. + +A controlled process test prints an execution proof but cannot apply a task disposition. The test operator completes that task through issue service. If the existing continuation policy already saved a disposition-repair retry, the test preserves it until its stored deadline. Only after all provider work has joined does it advance the clock; dispatch then revalidates the completed task and cancels the retry. The final quiet proof requires no registered deadlines, no tracked in-flight work and no owner SQL over twenty former polling minutes. It does not claim a saved future retry is already drained. diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts new file mode 100644 index 00000000000..4faac4ab535 --- /dev/null +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -0,0 +1,237 @@ +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issues, issueWatchdogs } from "@paperclipai/db"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { issueService } from "../services/issues.js"; +import { taskWatchdogService } from "../services/task-watchdogs.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_RECOVERY_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("issue recovery production producers and deadlines", () => { + let database: Awaited> | undefined; + let db: ReturnType; + let heartbeat: ReturnType; + let coordinator: ReturnType; + let worker: ReturnType["registerIssueStateRecoveryWork"]>; + let dispatch: ReturnType["registerRunDispatchWork"]>; + let cancellation: ReturnType["registerScheduledRetryCancellationWork"]>; + let draining = false; + const tasks = new Map number | null; hasPending: () => Promise }>(); + const errors: unknown[] = []; + const extraStops: Array<() => Promise> = []; + const extraDeadlines: Array<() => number | null> = []; + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("issue-recovery-work-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + }, 90_000); + beforeEach(async () => { + await db.execute(sql`TRUNCATE companies CASCADE`); + heartbeat = heartbeatService(db); draining = false; errors.length = 0; + }); + afterEach(async () => { + for (const stop of extraStops.splice(0)) await stop(); + extraDeadlines.length = 0; + await worker?.stop(); await dispatch?.stop(); await cancellation?.stop(); await coordinator?.stop(); + await heartbeat.drainActiveRunExecutions(); + vi.useRealTimers(); vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, + onError: error => errors.push(error) }); + tasks.clear(); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => { tasks.set(queue, task); return register(queue, task); }; + worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining); + dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining); + cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining); + await Promise.all([worker.ready, dispatch.ready, cancellation.ready]); + } + async function compete() { + const other = heartbeatService(db); + const competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const work = other.registerIssueStateRecoveryWork(competing, () => true); + extraStops.push(() => competing.stop()); extraDeadlines.push(competing.nextWakeAt); await work.ready; + } + async function fixture() { + const companyId = randomUUID(); + await db.insert(authUsers).values({ id: "recovery-user", name: "Recovery operator", email: "recovery-user@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); + await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `R${companyId.slice(0, 7)}`, + requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "recovery-user" }); + await db.insert(companyMemberships).values({ companyId, principalId: "recovery-user", principalType: "user", status: "active", membershipRole: "owner" }); + const [agent] = await db.insert(agents).values({ companyId, name: "Recovery process", role: "engineer", status: "idle", + adapterType: "process", adapterConfig: { command: process.execPath, args: ["-e", "console.log('recovery-proof')"], cwd: process.cwd() }, + runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } }, + }).returning(); + return { companyId, agentId: agent.id }; + } + async function settled() { + await vi.waitFor(async () => { + await heartbeat.drainActiveRunExecutions(); + const wake = coordinator.nextWakeAt(); + const detail = wake === null ? "" : JSON.stringify({ now: Date.now(), tasks: await Promise.all([...tasks].map(async ([queue, task]) => ({ queue, at: task.nextRunAt?.(), pending: await task.hasPending() }))), + runs: (await db.select().from(heartbeatRuns)).map(run => ({ id: run.id, status: run.status, error: run.errorCode, lease: run.controllerLeaseExpiresAt, reason: run.contextSnapshot?.wakeReason })) }); + expect(wake, detail).toBeNull(); + expect(extraDeadlines.map(next => next())).toEqual(extraDeadlines.map(() => null)); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + } + async function quiet() { + await settled(); + const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute"); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + try { await vi.advanceTimersByTimeAsync(20 * 60_000); expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); } + finally { vi.useRealTimers(); reads.mockRestore(); executes.mockRestore(); } + } + async function finish(issueId: string) { + await issueService(db).update(issueId, { status: "done" }); + await heartbeat.drainActiveRunExecutions(); + const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); + if (saved.length > 0) { + const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); + expect(await db.select().from(heartbeatRuns).where(and( + sql`${heartbeatRuns.status} in ('queued', 'running')`, + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`))).toHaveLength(0); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1); + await vi.waitFor(async () => { + const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); + expect(remaining).toHaveLength(0); + }); + for (const retry of saved) { + const [settledRetry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retry.id)); + expect(settledRetry.status).toBe("cancelled"); + } + } + } + async function completed(issueId: string) { + let runId: string | undefined; + await vi.waitFor(async () => { + const runs = await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`); + const run = runs.find(row => row.status === "running" || row.status === "succeeded"); + expect(run).toBeDefined(); runId = run!.id; + }, { timeout: 15_000 }); + await heartbeat.drainActiveRunExecutions(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!)); + expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode, stderr: run.stderrExcerpt, result: run.resultJson })).toBe("succeeded"); expect(run.stdoutExcerpt).toContain("recovery-proof"); + return run; + } + it("keeps empty, configured idle, and terminal history quiet", async () => { + await fixture(); await start(); await quiet(); + }); + it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => { + const f = await fixture(); await start(); await quiet(); + const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId }); + await completed(issue.id); + await finish(issue.id); await quiet(); + }, 45_000); + it("reacts to blocker resolution once and dispatches the durable dependency wake", async () => { + const f = await fixture(); + const blocker = await issueService(db).create(f.companyId, { title: "Blocker", status: "todo" }); + const dependent = await issueService(db).create(f.companyId, { title: "Dependent", status: "blocked", assigneeAgentId: f.agentId, + blockedByIssueIds: [blocker.id] }); + await start(); await quiet(); + await compete(); + await issueService(db).update(blocker.id, { status: "done" }); + const run = await completed(dependent.id); + expect(run.contextSnapshot?.wakeReason).toBe("issue_blockers_resolved"); + await finish(dependent.id); await quiet(); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'issue_blockers_resolved'`))).toHaveLength(1); + }, 45_000); + it("uses ordinary watchdog configuration and subtree closure to dispatch a reusable review", async () => { + const f = await fixture(); + const source = await issueService(db).create(f.companyId, { title: "Finished subtree", status: "done" }); + await start(); await quiet(); + await compete(); + const watchdog = await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId }); + let reviewId: string | undefined; + await vi.waitFor(async () => { + const [row] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.id, watchdog.watchdog.id)); + expect(row.watchdogIssueId).toBeTruthy(); reviewId = row.watchdogIssueId!; + }, { timeout: 15_000 }); + await completed(reviewId!); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'task_watchdog_stopped_subtree'`))).toHaveLength(1); + await finish(reviewId!); + await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet(); + }, 45_000); + it("bounds ordinary writes to their company and reconstructs foreign stale debt after restart", async () => { + const selected = await fixture(); await start(); await quiet(); + const foreign = await fixture(); + const [run] = await db.insert(heartbeatRuns).values({ companyId: foreign.companyId, agentId: foreign.agentId, + status: "failed", finishedAt: new Date() }).returning(); + const [stale] = await db.insert(issues).values({ companyId: foreign.companyId, title: "Foreign stale lock", status: "done", + checkoutRunId: run.id }).returning(); + await issueService(db).create(selected.companyId, { title: "Selected change", status: "done" }); await quiet(); + expect((await db.select().from(issues).where(eq(issues.id, stale.id)))[0].checkoutRunId).toBe(run.id); + await worker.stop(); await dispatch.stop(); await cancellation.stop(); await coordinator.stop(); + heartbeat = heartbeatService(db); await start(); await quiet(); + expect((await db.select().from(issues).where(eq(issues.id, stale.id)))[0].checkoutRunId).toBeNull(); + }); + it("fences nested assignment writes until outer commit and ignores rollback", async () => { + const f = await fixture(); await start(); await quiet(); + let committedId = ""; + await db.transaction(async outer => { + await outer.transaction(async nested => { + const issue = await issueService(db).create(f.companyId, { title: "Nested commit", status: "todo", assigneeAgentId: f.agentId }, nested); + committedId = issue.id; + }); + await new Promise(resolve => setTimeout(resolve, 100)); + expect(await db.select().from(heartbeatRuns)).toHaveLength(0); + }); + await completed(committedId); await finish(committedId); await settled(); + await expect(db.transaction(async outer => { + await outer.transaction(async nested => { + await issueService(db).create(f.companyId, { title: "Nested rollback", status: "todo", assigneeAgentId: f.agentId }, nested); + }); + throw new Error("rollback control"); + })).rejects.toThrow("rollback control"); + await quiet(); + expect(await db.select().from(issues).where(eq(issues.title, "Nested rollback"))).toHaveLength(0); + }, 60_000); + it("retains the first committed assignment during drain and executes on resume", async () => { + const f = await fixture(); await start(); await quiet(); draining = true; + const issue = await issueService(db).create(f.companyId, { title: "Drained assignment", status: "todo", assigneeAgentId: f.agentId }); + await new Promise(resolve => setTimeout(resolve, 100)); + const reads = vi.spyOn(db, "select"); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + try { await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); } + finally { vi.useRealTimers(); reads.mockRestore(); } + expect(await db.select().from(heartbeatRuns)).toHaveLength(0); + draining = false; worker.wake(); dispatch.wake(); + await completed(issue.id); await finish(issue.id); await quiet(); + }, 45_000); + it("retains a fresh subtree's exact first-run grace then disarms on disable", async () => { + const f = await fixture(); + const source = await issueService(db).create(f.companyId, { title: "Fresh unassigned subtree", status: "todo" }); + await start(); await quiet(); + await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId }); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(source.createdAt.getTime() + 15_000), { timeout: 5_000 }); + expect(await db.select().from(issues).where(eq(issues.originKind, "task_watchdog"))).toHaveLength(0); + await taskWatchdogService(db).disableForIssue(f.companyId, source.id); await quiet(); + }); + it("waits for a live foreign controller's exact stored lease before clearing a terminal lock", async () => { + const f = await fixture(); + const source = await issueService(db).create(f.companyId, { title: "Remote owner", status: "done" }); + const lease = new Date(Date.now() + 5_000); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", + runtimeMode: "legacy", controllerBootId: randomUUID(), controllerLeaseExpiresAt: lease, + contextSnapshot: { issueId: source.id }, processPid: 2_000_000_000, startedAt: new Date() }).returning(); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, source.id)); + await start(); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(lease.getTime()), { timeout: 3_000 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("running"); + await vi.waitFor(async () => expect((await db.select().from(issues).where(eq(issues.id, source.id)))[0].executionRunId).toBeNull(), { timeout: 12_000 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded"); await quiet(); + }, 30_000); +}); diff --git a/server/src/__tests__/issue-recovery-work-lifecycle.test.ts b/server/src/__tests__/issue-recovery-work-lifecycle.test.ts new file mode 100644 index 00000000000..4d0647f4406 --- /dev/null +++ b/server/src/__tests__/issue-recovery-work-lifecycle.test.ts @@ -0,0 +1,99 @@ +import { PgDialect } from "drizzle-orm/pg-core"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { issueWatchdogs } from "@paperclipai/db"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js"; +import { registerIssueRecoveryWork } from "../services/recovery/work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const stops: Array<() => Promise> = []; +afterEach(async () => { for (const stop of stops.splice(0)) await stop(); vi.useRealTimers(); }); +function fixture() { + vi.useFakeTimers(); vi.setSystemTime(1_000_000); + const dialect = new PgDialect(); + const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }> }; + const execute = vi.fn(async (query: any) => { + if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }]; + throw new Error("unexpected SQL"); + }); + const load = vi.fn((table: unknown) => table === issueWatchdogs ? rows.watchdogs : rows.issues); + const db: any = installDatabaseWorkSignals({ execute, + select: () => ({ from: (table: unknown) => ({ where: async () => load(table) }) }), + transaction: async (callback: (tx: any) => Promise) => callback({ execute, select: db.select, transaction: db.transaction }), + }); + const state = { draining: false, standby: false, suppressed: false, deadline: null as number | null, fail: false, incomplete: false }; + const check = vi.fn(async (scope: { onDeferredDeadline(at: number): void }) => { + if (state.fail) throw new Error("database unavailable"); + if (state.deadline !== null && state.deadline > Date.now()) scope.onDeferredDeadline(state.deadline); + }); + const owner = { stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; }, + locks: check, watchdogs: async (scope: any) => { await check(scope); return { incomplete: state.incomplete ? 1 : 0 }; }, + schedulingSuppressed: async () => state.suppressed }; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !state.standby, + canReconcile: () => !state.standby, onError: vi.fn() }); + const worker = registerIssueRecoveryWork(db, coordinator, owner, () => !state.draining && !state.standby); + stops.push(() => coordinator.stop()); + return { db, rows, state, check, load, coordinator, worker, + notify: () => db.transaction((tx: any) => notifyIssueRecovery(tx, { companyId: "selected" })), + settle: () => vi.advanceTimersByTimeAsync(1_000) }; +} +describe("issue recovery obligation lifecycle", () => { + it("has no timer, SQL, or owner probe after empty reconstruction", async () => { + const f = fixture(); await f.worker.ready; + expect(f.coordinator.nextWakeAt()).toBeNull(); const loaded = f.load.mock.calls.length; + await vi.advanceTimersByTimeAsync(20 * 60_000); + expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled(); + }); + it("retains exact grace and lease deadlines without early owner probes", async () => { + const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); + f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.deadline = 1_015_000; + await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_015_000); + const checked = f.check.mock.calls.length; + await vi.advanceTimersByTimeAsync(14_999); expect(f.check).toHaveBeenCalledTimes(checked); + await vi.advanceTimersByTimeAsync(1); expect(f.check.mock.calls.length).toBe(checked * 2); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("releases settled producer holds during domain drain while coordinator remains open", async () => { + const f = fixture(); await f.worker.ready; f.state.draining = true; + f.rows.issues.push({ id: "issue", companyId: "selected" }); + await f.notify(); await f.settle(); const loaded = f.load.mock.calls.length; + await vi.advanceTimersByTimeAsync(90_000); + expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + f.state.draining = false; f.worker.wake(); await f.settle(); + expect(f.check).toHaveBeenCalled(); expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("retries concrete failed entries and then disarms without reloading the company", async () => { + const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); f.state.fail = true; + await f.worker.ready; const loaded = f.load.mock.calls.length; + expect(f.coordinator.nextWakeAt()).toBe(1_030_000); + await vi.advanceTimersByTimeAsync(29_999); const checked = f.check.mock.calls.length; + f.state.fail = false; await vi.advanceTimersByTimeAsync(1); + expect(f.check.mock.calls.length).toBe(checked * 2); expect(f.load).toHaveBeenCalledTimes(loaded); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("keeps an incomplete watchdog subtree as concrete retry debt", async () => { + const f = fixture(); f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.incomplete = true; + await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_030_000); + const loaded = f.load.mock.calls.length; f.state.incomplete = false; + await vi.advanceTimersByTimeAsync(30_000); + expect(f.check).toHaveBeenCalledTimes(2); expect(f.load).toHaveBeenCalledTimes(loaded); + expect(f.coordinator.nextWakeAt()).toBeNull(); + }); + it("keeps a concurrent producer dirty until the in-flight owner has completed", async () => { + const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); await f.worker.ready; + let release!: () => void; const held = new Promise(resolve => { release = resolve; }); + f.check.mockImplementationOnce(() => held); + await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await f.settle(); + expect(f.coordinator.nextWakeAt()).toBeNull(); + expect(f.check.mock.calls.length).toBeGreaterThan(3); + }); + it("retains startup work on standby and cancels future deadlines on shutdown", async () => { + const f = fixture(); f.state.standby = true; f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); + await f.worker.ready; await vi.advanceTimersByTimeAsync(90_000); expect(f.load).not.toHaveBeenCalled(); + f.state.standby = false; f.state.deadline = Date.now() + 60_000; f.worker.wake(); await f.settle(); + expect(f.check).toHaveBeenCalledOnce(); await f.worker.stop(); + await vi.advanceTimersByTimeAsync(120_000); expect(f.check).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index e00c8519df7..a26d94f50d4 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -134,6 +134,7 @@ const { })), reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), + registerIssueStateRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -927,6 +928,7 @@ describe("startServer feedback export wiring", () => { it.each([false, true])("registers retry and queued dispatch only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerIssueStateRecoveryWork).toHaveBeenCalledTimes(enabled ? 1 : 0); expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0); if (enabled) { const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0]; @@ -1334,11 +1336,11 @@ describe("startServer feedback export wiring", () => { await startServer(); await settleStartupHousekeeping(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); - expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(1); + expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled(); expect(intervalCallback).not.toBeNull(); intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); - expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); + expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled(); diff --git a/server/src/index.ts b/server/src/index.ts index cd8bb951946..0ebe5c3332c 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -109,7 +109,6 @@ import { createApp } from "./app.js"; import { initializeCloudRuntimeConfig } from "./services/cloud-runtime-config.js"; import { loadCloudBootstrapConfig, loadConfig, type Config } from "./config.js"; import { logger } from "./middleware/logger.js"; -import { createIncompleteTaskWatchdogScanReporter } from "./services/task-watchdog-diagnostics.js"; import { setStartupRecoveryPhase } from "./startup-recovery-state.js"; import { StartupRefusalError, @@ -973,9 +972,6 @@ async function startServerWithDatabaseTeardown( } }; const pluginWorkerManager = createPluginWorkerManager(); - const reportIncompleteWatchdogScans = createIncompleteTaskWatchdogScanReporter( - (details, message) => logger.warn(details, message), - ); const heartbeat = config.heartbeatSchedulerEnabled ? heartbeatService(db as any, { pluginWorkerManager }) : null; @@ -1796,47 +1792,10 @@ async function startServerWithDatabaseTeardown( "startup session-goal recovery resumed durable agent goals", ); } - const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); - if ( - reconciled.assignmentDispatched > 0 || - reconciled.dispatchRequeued > 0 || - reconciled.continuationRequeued > 0 || - reconciled.successfulRunHandoffEscalated > 0 || - reconciled.successfulRunHandoffRetried > 0 || - reconciled.escalated > 0 - ) { - logger.warn( - { ...reconciled }, - "startup heartbeat recovery changed assigned issue state", - ); - } - - const dependencyWakesReconciled = await heartbeat.reconcileResolvedDependencyWakes(); - if (dependencyWakesReconciled.healed > 0) { - logger.warn( - { ...dependencyWakesReconciled }, - "startup dependency-wake reconciliation restored task execution paths", - ); - } - - const taskWatchdogsReconciled = await heartbeat.reconcileTaskWatchdogs(); - reportIncompleteWatchdogScans(taskWatchdogsReconciled, "startup"); - if (taskWatchdogsReconciled.triggered > 0) { - logger.warn( - { ...taskWatchdogsReconciled }, - "startup task-watchdog reconciliation triggered watchdog work", - ); - } - const scanned = await heartbeat.scanSilentActiveRuns(); if (scanned.created > 0 || scanned.escalated > 0) { logger.warn({ ...scanned }, "startup active-run output watchdog created review work"); } - - const swept = await heartbeat.sweepStaleIssueLocks(); - if (swept.cleared > 0) { - logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks"); - } })().catch((err) => { logger.error({ err }, "startup heartbeat recovery failed"); throw err; @@ -1845,6 +1804,8 @@ async function startServerWithDatabaseTeardown( await startupHeartbeatRecovery; } + if (config.heartbeatSchedulerEnabled) await heartbeat.registerIssueStateRecoveryWork(deliveryWork, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerRunDispatchWork(deliveryWork, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork, @@ -2051,33 +2012,6 @@ async function startServerWithDatabaseTeardown( .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true }) .then(async () => { await heartbeat.resumeQueuedRuns(); - const reconciled = await heartbeat.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true }); - if ( - reconciled.assignmentDispatched > 0 || - reconciled.dispatchRequeued > 0 || - reconciled.continuationRequeued > 0 || - reconciled.successfulRunHandoffEscalated > 0 || - reconciled.successfulRunHandoffRetried > 0 || - reconciled.escalated > 0 - ) { - logger.warn( - { ...reconciled }, - "periodic heartbeat recovery changed assigned issue state", - ); - } - }) - .then(async () => { - const reconciled = await heartbeat.reconcileResolvedDependencyWakes(); - if (reconciled.healed > 0) { - logger.warn({ ...reconciled }, "periodic dependency-wake reconciliation restored task execution paths"); - } - }) - .then(async () => { - const reconciled = await heartbeat.reconcileTaskWatchdogs(); - reportIncompleteWatchdogScans(reconciled, "periodic"); - if (reconciled.triggered > 0) { - logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work"); - } }) .then(async () => { const scanned = await heartbeat.scanSilentActiveRuns(); @@ -2085,12 +2019,6 @@ async function startServerWithDatabaseTeardown( logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); } }) - .then(async () => { - const swept = await heartbeat.sweepStaleIssueLocks(); - if (swept.cleared > 0) { - logger.warn({ ...swept }, "periodic stale-lock sweeper cleared issue locks"); - } - }) .catch((err) => { logger.error({ err }, "periodic heartbeat recovery failed"); })); diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index af44a0355d8..a10c6ee5eb6 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared"; @@ -544,6 +545,7 @@ export async function resumeConversationForReset(db: Db, comment: typeof issueCo eq(issues.id, comment.issueId), eq(issues.companyId, comment.companyId), )).for("update"); if (!isConversation(issue) || comment.conversationSessionGeneration != null) return; + await notifyIssueRecovery(tx, { companyId: issue.companyId }); const released = await tx.update(issueTreeHolds).set({ status: "released", releasedAt: new Date(), updatedAt: new Date(), releasedByActorType: "user", releasedByUserId: comment.authorUserId, diff --git a/server/src/services/chat-channels/telegram/attachment-failure.ts b/server/src/services/chat-channels/telegram/attachment-failure.ts index 58d5d4a2677..785b3116baf 100644 --- a/server/src/services/chat-channels/telegram/attachment-failure.ts +++ b/server/src/services/chat-channels/telegram/attachment-failure.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "../../issue-recovery-notifications.js"; import { createProviderEffects, type ProviderEffectTarget } from "../provider-effects.js"; import type { EndpointRow, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js"; @@ -78,6 +79,7 @@ export function createTelegramAttachmentFailure(db: Db, options: TelegramAttachm })(), ); const effect = await db.transaction(async (tx) => { + await notifyIssueRecovery(tx, { companyId: input.endpoint.companyId }); // Replace the optimistic attachment placeholder with an explicit safe // Paperclip-visible failure. The message link stays authoritative for // exact provider redelivery, while no agent wakeup is enqueued for an diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 7ab88aadfa9..26bc4e081a1 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "./instruction-collection-notifications.js"; @@ -71,6 +72,7 @@ export async function deleteCompany(db: Db, id: string) { const [existing] = await tx.select({ id: companies.id }).from(companies) .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; + await notifyIssueRecovery(tx, { companyId: id }); await notifyChatCompanyConfiguration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); diff --git a/server/src/services/company-portability.ts b/server/src/services/company-portability.ts index dc7dca311d6..585b8fcf5f4 100644 --- a/server/src/services/company-portability.ts +++ b/server/src/services/company-portability.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { createAgentLifecycle } from "./agent-lifecycle.js"; import { agentAppearanceSchema, agentHarnessType, paperclipRunnerProfileForHarness, agentRunner, type AgentRunnerChoice } from "@paperclipai/shared"; import { resolveNewAgentRunnerForCompany } from "./agent-runner-selection.js"; @@ -6465,7 +6466,9 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { } if (relationRows.length > 0) { const relationCompanyId = targetCompany.id; - await db + await db.transaction(async tx => { + await notifyIssueRecovery(tx, { companyId: relationCompanyId }); + await tx .insert(issueRelations) .values(relationRows.map((row) => ({ companyId: relationCompanyId, @@ -6476,6 +6479,7 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { createdByUserId: actorUserId ?? null, }))) .onConflictDoNothing(); + }); } } diff --git a/server/src/services/documents.ts b/server/src/services/documents.ts index 493c495e007..814c68eeea7 100644 --- a/server/src/services/documents.ts +++ b/server/src/services/documents.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { randomUUID } from "node:crypto"; import { and, asc, desc, eq, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -275,6 +276,7 @@ export function documentService(db: Db) { .where(eq(issueDocuments.issueId, issue.id)); const fallbackKey = nextAvailableDocumentKey(key, issueDocumentKeys.map((row) => row.key)); + await notifyIssueRecovery(tx, { companyId: issue.companyId }); const [document] = await tx .insert(documents) .values({ @@ -376,6 +378,7 @@ export function documentService(db: Db) { }); } + await notifyIssueRecovery(tx, { companyId: issue.companyId }); const nextRevisionNumber = existing.latestRevisionNumber + 1; const [revision] = await tx .insert(documentRevisions) @@ -438,6 +441,7 @@ export function documentService(db: Db) { throw conflict("Document does not exist yet", { key }); } + await notifyIssueRecovery(tx, { companyId: issue.companyId }); const [document] = await tx .insert(documents) .values({ @@ -592,6 +596,9 @@ export function documentService(db: Db) { }); } await db.transaction(async (tx) => { + for (const companyId of new Set(rows.map(row => row.companyId))) { + await notifyIssueRecovery(tx, { companyId }); + } await insertRowsInChunks(tx, documents, documentRows); await insertRowsInChunks(tx, documentRevisions, revisionRows); await insertRowsInChunks(tx, issueDocuments, linkRows); @@ -644,6 +651,7 @@ export function documentService(db: Db) { }); } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); const now = new Date(); const nextRevisionNumber = existing.latestRevisionNumber + 1; const [restoredRevision] = await tx @@ -722,6 +730,7 @@ export function documentService(db: Db) { }; } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); const now = new Date(); await tx .update(documents) @@ -769,6 +778,7 @@ export function documentService(db: Db) { }; } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); const now = new Date(); await tx .update(documents) @@ -817,6 +827,7 @@ export function documentService(db: Db) { }); } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); await tx.delete(issueDocuments).where(eq(issueDocuments.documentId, existing.id)); await tx.delete(documents).where(eq(documents.id, existing.id)); diff --git a/server/src/services/external-conversation-settlement-notifications.ts b/server/src/services/external-conversation-settlement-notifications.ts index 46f4493ff2a..d5e4cdd506c 100644 --- a/server/src/services/external-conversation-settlement-notifications.ts +++ b/server/src/services/external-conversation-settlement-notifications.ts @@ -1,7 +1,9 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import type { ExternalConversationSettlementScope } from "./slack-conversation-lifecycle.js"; -export function notifyExternalConversationSettlement(tx: object, scope?: ExternalConversationSettlementScope) { +export async function notifyExternalConversationSettlement(tx: object, scope?: ExternalConversationSettlementScope) { + await notifyIssueRecovery(tx, scope); return notifyDeliveryWork(tx, DELIVERY_QUEUES.externalConversationSettlement, scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*"); } diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index 48fb001c0b4..316eb413afe 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; import { writeExecutionControl } from "./execution-control-notifications.js"; @@ -322,6 +323,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { ): Promise { try { const created = await db.transaction(async (tx) => { + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition); await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); @@ -413,6 +415,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { Math.min(nextAttemptCount, effectiveMaxAttempts), ); const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); if ((!input.preserveExistingOwner && existing.cause !== input.cause) || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) { @@ -474,6 +477,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { return toReadModel(exhausted); } const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); if ((!input.preserveExistingOwner && existing.cause !== input.cause) || (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) { @@ -548,6 +552,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { try { const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => { + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); return tx @@ -595,6 +600,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { const pending = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions) .where(and(...predicates)).limit(1); if (!pending.length) return []; + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId }); return tx diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 0d7633f846a..ff558b81ef0 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js"; @@ -758,6 +759,7 @@ export function issueTreeControlService(db: Db) { const releaseReason = input.reason ?? "Subtree resume applied."; const { hold: resumeHold } = await db.transaction(async (tx) => { + await notifyIssueRecovery(tx, { companyId }); const [createdHold] = await tx .insert(issueTreeHolds) .values({ @@ -836,6 +838,7 @@ export function issueTreeControlService(db: Db) { } const { hold, members } = await db.transaction(async (tx) => { + await notifyIssueRecovery(tx, { companyId }); const [createdHold] = await tx .insert(issueTreeHolds) .values({ @@ -1177,7 +1180,9 @@ export function issueTreeControlService(db: Db) { throw conflict("Issue tree hold is already released"); } - const [updated] = await db + const [updated] = await db.transaction(async tx => { + await notifyIssueRecovery(tx, { companyId }); + return tx .update(issueTreeHolds) .set({ status: "released", @@ -1195,6 +1200,7 @@ export function issueTreeControlService(db: Db) { }) .where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId))) .returning(); + }); const members = await db .select() diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 575648ccc37..5a569f4ce1e 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { notifyRunDispatch } from "./run-dispatch-notifications.js"; import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js"; import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js"; @@ -7627,6 +7628,7 @@ export function issueService(db: Db) { await assertNoBlockingCycles(companyId, issueId, deduped, dbOrTx); } + await notifyIssueRecovery(dbOrTx, { companyId }); await dbOrTx .delete(issueRelations) .where( @@ -13799,6 +13801,7 @@ export function issueService(db: Db) { .then((rows) => rows[0] ?? null); if (!existing) return null; + await notifyIssueRecovery(tx, { companyId: existing.companyId }); await tx.delete(issueAttachments).where(eq(issueAttachments.id, id)); await tx.delete(assets).where(eq(assets.id, existing.assetId)); return existing; diff --git a/server/src/services/native-runtime/chat-attachment-reuse.ts b/server/src/services/native-runtime/chat-attachment-reuse.ts index cc7990e9b7a..6cc44299ee7 100644 --- a/server/src/services/native-runtime/chat-attachment-reuse.ts +++ b/server/src/services/native-runtime/chat-attachment-reuse.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "../issue-recovery-notifications.js"; import { createHash } from "node:crypto"; import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; @@ -1582,7 +1583,9 @@ export async function prepareReusedChatAttachment(input: { if (!workProduct) { throw new Error("paperclip_runner_chat_attachment_origin_not_persisted"); } - await input.db + await input.db.transaction(async tx => { + await notifyIssueRecovery(tx, { companyId: input.binding.companyId }); + await tx .update(issueWorkProducts) .set({ title: input.title, @@ -1602,6 +1605,7 @@ export async function prepareReusedChatAttachment(input: { eq(issueWorkProducts.createdByRunId, input.binding.runId), ), ); + }); const comment = await issueService(input.db).addComment( input.binding.issueId, `Prepared ${input.title} for this response.`, diff --git a/server/src/services/native-runtime/native-runner-file-handoff.ts b/server/src/services/native-runtime/native-runner-file-handoff.ts index 6f2c32beec8..420380500ea 100644 --- a/server/src/services/native-runtime/native-runner-file-handoff.ts +++ b/server/src/services/native-runtime/native-runner-file-handoff.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "../issue-recovery-notifications.js"; import { createHash, randomUUID } from "node:crypto"; import { execFile } from "node:child_process"; import { constants } from "node:fs"; @@ -1265,7 +1266,9 @@ export async function prepareNativeRunnerFileHandoff(input: { ) { throw new Error("paperclip_runner_file_handoff_origin_not_persisted"); } - await input.db + await input.db.transaction(async tx => { + await notifyIssueRecovery(tx, { companyId: input.binding.companyId }); + await tx .update(issueWorkProducts) .set({ title: verified.title, updatedAt: new Date() }) .where( @@ -1276,6 +1279,7 @@ export async function prepareNativeRunnerFileHandoff(input: { eq(issueWorkProducts.createdByRunId, input.binding.runId), ), ); + }); const comment = await issueService(input.db).addComment( input.binding.issueId, `Prepared ${verified.title} for this response.`, diff --git a/server/src/services/native-status-reassessment-notifications.ts b/server/src/services/native-status-reassessment-notifications.ts index fb845523bb4..4b5eac73b2f 100644 --- a/server/src/services/native-status-reassessment-notifications.ts +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -1,8 +1,10 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; -export function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { +export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { + await notifyIssueRecovery(tx, scope); return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); } diff --git a/server/src/services/system-recovery.ts b/server/src/services/system-recovery.ts index a554a765c4c..347593dafed 100644 --- a/server/src/services/system-recovery.ts +++ b/server/src/services/system-recovery.ts @@ -1,3 +1,4 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { and, eq, sql } from "drizzle-orm"; import { heartbeatRuns, issueComments, issueRecoveryActions, type Db } from "@paperclipai/db"; import { writeExecutionControl } from "./execution-control-notifications.js"; @@ -11,6 +12,7 @@ export async function adoptSystemRecovery(db: Db, source: WorkspaceRecoverySourc .where(and(eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.id, source.actionId), eq(issueRecoveryActions.sourceIssueId, source.issueId))).for("update"); if (!existing || existing.evidence.systemRecovery) return; + await notifyIssueRecovery(tx, { companyId: source.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); const state = createSystemRecovery(source.runId); // Atomic, once per source. An exhausted previous execution budget is history; diff --git a/server/src/services/task-watchdog-work.ts b/server/src/services/task-watchdog-work.ts index 091c8c6595f..cde0a7eca43 100644 --- a/server/src/services/task-watchdog-work.ts +++ b/server/src/services/task-watchdog-work.ts @@ -29,7 +29,7 @@ export function registerTaskWatchdogWork( let next: number | null = null; const result = await owner.reconcile({ companyId: entry.companyId, watchdogId: entry.id, onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } }); - if (result.incomplete > 0) throw new Error("Task watchdog subtree remains incomplete"); + if (result.incomplete > 0) throw new Error(`Task watchdog ${entry.id} in company ${entry.companyId} has an incomplete subtree`); return next === null ? null : { ...entry, dueAt: next }; }, }, canRun); diff --git a/server/src/services/task-watchdogs.ts b/server/src/services/task-watchdogs.ts index 6f94e5520a7..dd1aa980d3c 100644 --- a/server/src/services/task-watchdogs.ts +++ b/server/src/services/task-watchdogs.ts @@ -848,6 +848,7 @@ async function upsertIssueWatchdogInTransaction( .where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId))) .then((rows: IssueWatchdogRow[]) => rows[0] ?? null); + await notifyIssueRecovery(dbOrTx, { companyId }); if (existing) { const updated = await updateIssueWatchdogRow(dbOrTx, existing, input, now); return { watchdog: toIssueWatchdog(updated), created: false }; @@ -892,10 +893,7 @@ export function upsertIssueWatchdogForIssue( issueId: string, input: IssueWatchdogUpsertInput, ): Promise<{ watchdog: IssueWatchdog; created: boolean }> { - return dbOrTx.transaction(async tx => { - await notifyIssueRecovery(tx, { companyId }); - return upsertIssueWatchdogInTransaction(tx, companyId, issueId, input); - }); + return dbOrTx.transaction(tx => upsertIssueWatchdogInTransaction(tx, companyId, issueId, input)); } export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) {