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 9c9be8f00af..e4a9f3fd922 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -137,6 +137,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() })), @@ -980,6 +981,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]; @@ -1393,11 +1395,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 645f7affcd2..1a7106ad0b5 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -112,7 +112,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, @@ -974,9 +973,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; @@ -1748,47 +1744,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; @@ -1797,6 +1756,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, @@ -1957,33 +1918,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(); @@ -1991,12 +1925,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 6dc783568e2..ef8a5e554f7 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -1,4 +1,5 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; +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"; @@ -547,6 +548,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 49e2497a055..a7e9e4367e3 100644 --- a/server/src/services/chat-channels/telegram/attachment-failure.ts +++ b/server/src/services/chat-channels/telegram/attachment-failure.ts @@ -1,4 +1,5 @@ import { notifyStatusCardIssueChange } from "../../status-card-notifications.js"; +import { notifyIssueRecovery } from "../../issue-recovery-notifications.js"; import { createProviderEffects, type ProviderEffectTarget } from "../provider-effects.js"; import type { EndpointRow, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js"; @@ -79,6 +80,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 c2e92ee16f7..66e81df7673 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,5 +1,6 @@ import { notifyStatusCardEvaluation } from "./status-card-notifications.js"; import { notifyExternalObjectRefresh } from "./external-object-refresh-notifications.js"; +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"; @@ -74,6 +75,7 @@ export async function deleteCompany(db: Db, id: string) { .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; await notifyExternalObjectRefresh(tx, id); + 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 fff45394e08..0b41482b0fa 100644 --- a/server/src/services/company-portability.ts +++ b/server/src/services/company-portability.ts @@ -1,4 +1,5 @@ import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +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"; @@ -6468,7 +6469,8 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { const relationCompanyId = targetCompany.id; await db.transaction(async tx => { await notifyAttentionRetentionWork(tx, { companyId: relationCompanyId }); - return tx + await notifyIssueRecovery(tx, { companyId: relationCompanyId }); + await tx .insert(issueRelations) .values(relationRows.map((row) => ({ companyId: relationCompanyId, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index d9e9f25934a..a0caf0bf3e8 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,10 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + strandedAssignments: "stranded-assignments", + resolvedDependencies: "resolved-dependencies", + taskWatchdogs: "task-watchdogs", + staleIssueLocks: "stale-issue-locks", scheduledRetryDispatch: "scheduled-retry-dispatch", queuedRunDispatch: "queued-run-dispatch", statusCardEvaluation: "status-card-evaluation", diff --git a/server/src/services/documents.ts b/server/src/services/documents.ts index 79e8e14309c..7c81d075354 100644 --- a/server/src/services/documents.ts +++ b/server/src/services/documents.ts @@ -1,4 +1,5 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; +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"; @@ -277,6 +278,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({ @@ -378,6 +380,7 @@ export function documentService(db: Db) { }); } + await notifyIssueRecovery(tx, { companyId: issue.companyId }); const nextRevisionNumber = existing.latestRevisionNumber + 1; const [revision] = await tx .insert(documentRevisions) @@ -440,6 +443,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({ @@ -594,7 +598,10 @@ export function documentService(db: Db) { }); } await db.transaction(async (tx) => { - for (const companyId of new Set(rows.map(row => row.companyId))) await notifyStatusCardIssueChange(tx, companyId); + for (const companyId of new Set(rows.map(row => row.companyId))) { + await notifyStatusCardIssueChange(tx, companyId); + await notifyIssueRecovery(tx, { companyId }); + } await insertRowsInChunks(tx, documents, documentRows); await insertRowsInChunks(tx, documentRevisions, revisionRows); await insertRowsInChunks(tx, issueDocuments, linkRows); @@ -647,6 +654,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 @@ -726,6 +734,7 @@ export function documentService(db: Db) { }; } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); const now = new Date(); await tx .update(documents) @@ -773,6 +782,7 @@ export function documentService(db: Db) { }; } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); const now = new Date(); await tx .update(documents) @@ -821,6 +831,7 @@ export function documentService(db: Db) { }); } + await notifyIssueRecovery(tx, { companyId: existing.companyId }); await tx.delete(issueDocuments).where(eq(issueDocuments.documentId, existing.id)); await notifyStatusCardIssueChange(tx, existing.companyId); 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/heartbeat.ts b/server/src/services/heartbeat.ts index 2160f451b23..42dc9928ad5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerIssueRecoveryWork } from "./recovery/work.js"; import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js"; import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js"; import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js"; @@ -1611,15 +1612,15 @@ export function heartbeatService( void reportRunFailure(db, updated, failureReport); } - async function reconcileStrandedAssignedIssues(opts?: { skipExternalConversationSettlement?: boolean }) { + async function reconcileStrandedAssignedIssues(opts?: Parameters[0]) { return recovery.reconcileStrandedAssignedIssues({ - skipExternalConversationSettlement: opts?.skipExternalConversationSettlement, + ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff(), }); } - async function sweepStaleIssueLocks() { - return recovery.sweepStaleIssueLocks(); + async function sweepStaleIssueLocks(opts?: Parameters[0]) { + return recovery.sweepStaleIssueLocks(opts); } function issueIdFromRunContext(contextSnapshot: unknown) { @@ -1649,10 +1650,7 @@ export function heartbeatService( }); } - async function reconcileTaskWatchdogs(opts?: { - companyId?: string | null; - runId?: string | null; - }) { + async function reconcileTaskWatchdogs(opts?: Parameters[0]) { return taskWatchdogs.reconcileTaskWatchdogs({ ...opts, issueCreatedAtGte: await getWorktreeExecutionCutoff(), @@ -1677,10 +1675,7 @@ export function heartbeatService( return recovery.buildRunOutputSilence(run, now); } - async function reconcileResolvedDependencyWakes(opts?: { - runId?: string | null; - companyId?: string | null; - }) { + async function reconcileResolvedDependencyWakes(opts?: Parameters[0]) { return recovery.reconcileResolvedDependencyWakeBackstop(opts); } @@ -4248,6 +4243,14 @@ export function heartbeatService( computeTaskDrain, applyTaskDrain, + registerIssueStateRecoveryWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerIssueRecoveryWork(db, coordinator, { + stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true }), + dependencies: reconcileResolvedDependencyWakes, + locks: sweepStaleIssueLocks, + watchdogs: reconcileTaskWatchdogs, + schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed, + }, () => canRun() && readTaskDrain(new Date()) === null), registerRunDispatchWork: (coordinator: Parameters[0], canRun: () => boolean) => registerRunDispatchWork(coordinator, { promoteDueScheduledRetries, diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts index 4d4cf374e5b..cff04619945 100644 --- a/server/src/services/issue-comment-writes.ts +++ b/server/src/services/issue-comment-writes.ts @@ -1,4 +1,5 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import type { PgInsertValue } from "drizzle-orm/pg-core"; import { issueComments, type Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -8,6 +9,7 @@ type WriteDb = Db | Parameters[0]>[0]; export function insertIssueComment(db: WriteDb, input: PgInsertValue & { companyId: string }) { return db.transaction(async tx => { await notifyStatusCardIssueChange(tx, input.companyId); + await notifyIssueRecovery(tx, { companyId: input.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`); return tx.insert(issueComments).values(input).returning(); }); diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index a1ccb8e8e7a..e276dc88f89 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -1,4 +1,5 @@ import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +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"; @@ -324,6 +325,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) { try { const created = await db.transaction(async (tx) => { await notifyAttentionRetentionWork(tx, { companyId: input.companyId }); + 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); @@ -415,6 +417,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)) { @@ -476,6 +479,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)) { @@ -550,6 +554,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 @@ -597,6 +602,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-recovery-notifications.ts b/server/src/services/issue-recovery-notifications.ts new file mode 100644 index 00000000000..db04ca801ba --- /dev/null +++ b/server/src/services/issue-recovery-notifications.ts @@ -0,0 +1,11 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type IssueRecoveryScope = { companyId: string }; + +export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) { + const hint = scope ? JSON.stringify(scope) : "*"; + for (const queue of [DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, + DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) { + await notifyDeliveryWork(transaction, queue, hint); + } +} diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index e784be722a5..c82c8da7485 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,5 +1,6 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js"; +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"; @@ -760,6 +761,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({ @@ -838,6 +840,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({ @@ -1181,7 +1184,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", @@ -1199,6 +1204,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 22642b3fcc9..fb2de0de57b 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 { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; @@ -7634,6 +7635,7 @@ export function issueService(db: Db) { } await notifyAttentionRetentionWork(dbOrTx, { companyId }); + await notifyIssueRecovery(dbOrTx, { companyId }); await dbOrTx .delete(issueRelations) .where( @@ -11521,7 +11523,9 @@ export function issueService(db: Db) { await notifyMergeConfirmationWork(tx, { companyId: receiptExisting.companyId, issueId: id }); } if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined - || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) { + || patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined || patch.parentId !== undefined + || patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined + || blockedByIssueIds !== undefined) { await notifyRunDispatch(tx, { companyId: receiptExisting.companyId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); await notifyRoutineExecutionStatus(tx, id); @@ -13852,6 +13856,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/legacy-controller-lease.ts b/server/src/services/legacy-controller-lease.ts index 55a8439debc..22fa078f459 100644 --- a/server/src/services/legacy-controller-lease.ts +++ b/server/src/services/legacy-controller-lease.ts @@ -77,14 +77,18 @@ async function renewOnLivenessPool( }); } -export async function hasLiveLegacyController(db: Db, run: Run): Promise { - if (run.runtimeMode === "native" || !run.controllerBootId) return false; - const [owner] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( +export async function liveLegacyControllerExpiresAt(db: Db, run: Run): Promise { + if (run.runtimeMode === "native" || !run.controllerBootId) return null; + const [owner] = await db.select({ expiresAt: heartbeatRuns.controllerLeaseExpiresAt }).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.status, "running"), gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), )); - return Boolean(owner); + return owner?.expiresAt ?? null; +} + +export async function hasLiveLegacyController(db: Db, run: Run): Promise { + return (await liveLegacyControllerExpiresAt(db, run)) !== null; } export type RevokedLegacyControllerState = Pick; diff --git a/server/src/services/native-runtime/chat-attachment-reuse.test.ts b/server/src/services/native-runtime/chat-attachment-reuse.test.ts index ccd4bbe30d6..d0de5ab04c9 100644 --- a/server/src/services/native-runtime/chat-attachment-reuse.test.ts +++ b/server/src/services/native-runtime/chat-attachment-reuse.test.ts @@ -73,7 +73,7 @@ describe("native same-conversation chat attachment reuse", () => { beforeAll(async () => { const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CHAT_REUSE_TEST_DATABASE_URL"); if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-chat-reuse-"); - db = createDb(external ?? temporary!.connectionString); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); const storageRoot = await mkdtemp( path.join(tmpdir(), "paperclip-chat-reuse-"), ); 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.test.ts b/server/src/services/native-runtime/native-runner-file-handoff.test.ts index 0abae30539f..e302c432190 100644 --- a/server/src/services/native-runtime/native-runner-file-handoff.test.ts +++ b/server/src/services/native-runtime/native-runner-file-handoff.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js"; import { nativeCompletionFeedback } from "./native-completion-feedback.js"; import { createHash, randomUUID } from "node:crypto"; @@ -61,10 +62,9 @@ describe("native runner file handoff", () => { const runId = "00000000-0000-4000-8000-000000009104"; beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase( - "native-runner-file-handoff-", - ); - db = createDb(temporary.connectionString); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_HANDOFF_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-runner-file-handoff-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 }); temporaryRoot = await mkdtemp( path.join(tmpdir(), "paperclip-native-file-handoff-"), ); 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 6989821ee90..94f7597e67d 100644 --- a/server/src/services/native-status-reassessment-notifications.ts +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -1,5 +1,6 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; import { notifyAttentionRetentionWork } from "./retention-work-notifications.js"; +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 }; @@ -7,6 +8,7 @@ export type NativeStatusReassessmentScope = { companyId: string; issueId?: strin export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { await notifyStatusCardIssueChange(tx, scope.companyId); await notifyAttentionRetentionWork(tx, { companyId: scope.companyId }); + 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/recovery/service.ts b/server/src/services/recovery/service.ts index fffc2699dbf..7888b3e23fb 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -27,7 +27,7 @@ import { decideLegacyContinuation, legacyDispositionEpisode, legacyDispositionFingerprint, LEGACY_DISPOSITION_REPAIR_INSTRUCTION, type LegacyDispositionEpisode, } from "./legacy-continuation.js"; -import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController } from "../legacy-controller-lease.js"; +import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController, liveLegacyControllerExpiresAt } from "../legacy-controller-lease.js"; import { isCompletedOnboardingHandoffWake } from "../chat-completion-delivery.js"; import { instanceSettingsService } from "../instance-settings.js"; import { isWaitingConversation, settleConversationTurnForRun, deliverConversationComments } from "../agent-conversations.js"; @@ -245,6 +245,7 @@ type ResolvedDependencyWakeBackstopOptions = { runId?: string | null; companyId?: string | null; blockerIssueId?: string | null; + issueId?: string; source?: ResolvedDependencyWakeBackstopSource; }; @@ -4526,6 +4527,9 @@ export function recoveryService( } async function reconcileStrandedAssignedIssues(opts?: { + companyId?: string; + issueId?: string; + onDeferredDeadline?: (at: number) => void; issueCreatedAtGte?: Date | null; skipExternalConversationSettlement?: boolean; }) { @@ -4534,6 +4538,8 @@ export function recoveryService( .from(issues) .where( and( + opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined, + opts?.issueId ? eq(issues.id, opts.issueId) : undefined, isNull(issues.assigneeUserId), inArray(issues.status, ["todo", "in_progress", "in_review"]), or( @@ -5714,6 +5720,7 @@ export function recoveryService( classification.baseBackoffMs * Math.pow(2, Math.max(0, consecutive - 1)); if (elapsed < requiredDelay) { + opts?.onDeferredDeadline?.(latestFinishedAt.getTime() + requiredDelay); result.skipped += 1; continue; } @@ -5815,7 +5822,7 @@ export function recoveryService( source === "workspace.finalize" ? "workspace_finalize_reconciliation" : "issue_graph_liveness_reconciliation"; - const useCursor = !opts?.blockerIssueId; + const useCursor = !opts?.blockerIssueId && !opts?.issueId; const queryCandidates = (afterIssueId: string | null) => { const filters = [ @@ -5825,6 +5832,7 @@ export function recoveryService( sql`${issues.assigneeAgentId} is not null`, ]; if (opts?.companyId) filters.push(eq(issues.companyId, opts.companyId)); + if (opts?.issueId) filters.push(eq(issues.id, opts.issueId)); if (afterIssueId) filters.push(gt(issues.id, afterIssueId)); if (opts?.blockerIssueId) { @@ -6115,6 +6123,7 @@ export function recoveryService( // fallback below. Without it, a terminal issue named in the run context // snapshot would still terminalize the shared run and defeat the guard. runReferencedByActiveIssue?: boolean; + onDeferredDeadline?: (at: number) => void; }, ): Promise<{ terminalized: boolean; status: string }> { // Act only on a run in "running" status. A "queued" run has no process yet, @@ -6130,8 +6139,11 @@ export function recoveryService( // Another controller may own a sandbox run whose PID has no meaning on // this host. Its live lease owns both execution and finalization, even if // the issue is already terminal or this process has no in-memory handle. - if (await hasLiveLegacyController(db, run)) + const controllerDeadline = await liveLegacyControllerExpiresAt(db, run); + if (controllerDeadline) { + options?.onDeferredDeadline?.(controllerDeadline.getTime()); return { terminalized: false, status: run.status }; + } const pid = run.processPid ?? null; const processGroupId = run.processGroupId ?? null; @@ -6352,7 +6364,7 @@ export function recoveryService( // still claims to be live but can no longer reach a terminal status on its // own, so a stuck "running" run can no longer block the sweep. Idempotent and // safe: clears at most one row's worth of lock columns per candidate. - async function sweepStaleIssueLocks() { + async function sweepStaleIssueLocks(opts?: { companyId?: string; issueId?: string; onDeferredDeadline?: (at: number) => void }) { const result = { cleared: 0, issueIds: [] as string[], @@ -6369,7 +6381,9 @@ export function recoveryService( }) .from(issues) .where( - sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`, + and(sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`, + opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined, + opts?.issueId ? eq(issues.id, opts.issueId) : undefined), ); const referencedRunIds = [ @@ -6384,7 +6398,8 @@ export function recoveryService( ? await db .select() .from(heartbeatRuns) - .where(inArray(heartbeatRuns.id, referencedRunIds)) + .where(and(inArray(heartbeatRuns.id, referencedRunIds), + opts?.companyId ? eq(heartbeatRuns.companyId, opts.companyId) : undefined)) : []; const runStatusById = new Map(); for (const row of runRows) runStatusById.set(row.id, row.status); @@ -6395,8 +6410,15 @@ export function recoveryService( // the same run id in a stale lock column. The terminal reference alone must // not terminalize a run that an active issue still owns, so exclude these // runs from the issue-terminal authority below. + const ownershipReferences = opts?.issueId && referencedRunIds.length > 0 + ? await db.select({ companyId: issues.companyId, status: issues.status, + checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }).from(issues).where(and( + opts.companyId ? eq(issues.companyId, opts.companyId) : undefined, + or(inArray(issues.checkoutRunId, referencedRunIds), inArray(issues.executionRunId, referencedRunIds)), + )) + : candidates; const runIdsReferencedByActiveIssue = new Set(); - for (const issue of candidates) { + for (const issue of ownershipReferences) { if (issue.status === "done" || issue.status === "cancelled") continue; for (const runId of [issue.checkoutRunId, issue.executionRunId]) { if (runId) runIdsReferencedByActiveIssue.add(runId); @@ -6437,6 +6459,7 @@ export function recoveryService( referencingIssueTerminalStatus: issueTerminalStatusByRunId.get(row.id) ?? null, runReferencedByActiveIssue: runIdsReferencedByActiveIssue.has(row.id), + onDeferredDeadline: opts?.onDeferredDeadline, }); runStatusById.set(row.id, outcome.status); if (outcome.terminalized) result.terminalizedRunIds.push(row.id); diff --git a/server/src/services/recovery/work.ts b/server/src/services/recovery/work.ts new file mode 100644 index 00000000000..93679e80423 --- /dev/null +++ b/server/src/services/recovery/work.ts @@ -0,0 +1,58 @@ +import { registerTaskWatchdogWork } from "../task-watchdog-work.js"; +import { and, eq, inArray, isNotNull, isNull, or } from "drizzle-orm"; +import { issues, 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 { IssueRecoveryScope } from "../issue-recovery-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; + +type Entry = { id: string; companyId: string; dueAt: number }; +type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void }; + +export function registerIssueRecoveryWork( + db: Db, + coordinator: ReturnType, + owner: { + stranded(scope: Scope): Promise; + dependencies(scope: Scope): Promise<{ enqueueFailed: number }>; + locks(scope: Scope): Promise; + watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>; + schedulingSuppressed(): Promise; + }, + canRun: () => boolean, +) { + function register(queue: DeliveryQueue, statuses: string[] | null, check: (scope: Scope) => Promise) { + return registerScopedDeadlineWork(coordinator, { + queue, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, + matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + load: async scope => { + const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and( + scope ? eq(issues.companyId, scope.companyId) : undefined, + statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)), + statuses ? isNull(issues.hiddenAt) : undefined, + )); + return rows.map(row => ({ ...row, dueAt: Date.now() })); + }, + check: async entry => { + if (await owner.schedulingSuppressed()) return null; + let next: number | null = null; + await check({ companyId: entry.companyId, issueId: entry.id, + onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } }); + return next === null ? null : { ...entry, dueAt: next }; + }, + }, canRun); + } + const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded); + const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => { + const result = await owner.dependencies(scope); + if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered"); + }); + const locks = register(DELIVERY_QUEUES.staleIssueLocks, null, owner.locks); + const watchdogs = registerTaskWatchdogWork(db, coordinator, { + reconcile: owner.watchdogs, enabled: async () => !(await owner.schedulingSuppressed()), + }, canRun); + return { ready: Promise.all([assignments.ready, dependencies.ready, locks.ready, watchdogs.ready]), + wake() { assignments.invalidate(); dependencies.invalidate(); locks.invalidate(); watchdogs.invalidate(); }, + async stop() { await Promise.all([assignments.stop(), dependencies.stop(), locks.stop(), watchdogs.stop()]); } }; +} diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 2e04e6e7531..f725add167f 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -1,8 +1,10 @@ +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { + await notifyIssueRecovery(transaction, scope); const hint = scope ? JSON.stringify(scope) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); 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 new file mode 100644 index 00000000000..cde0a7eca43 --- /dev/null +++ b/server/src/services/task-watchdog-work.ts @@ -0,0 +1,36 @@ +import { and, eq } from "drizzle-orm"; +import { issueWatchdogs, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { IssueRecoveryScope } from "./issue-recovery-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; + +type Entry = { id: string; companyId: string; dueAt: number }; + +export function registerTaskWatchdogWork( + db: Db, + coordinator: ReturnType, + owner: { + enabled(): Promise; + reconcile(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>; + }, + canRun: () => boolean, +) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.taskWatchdogs, + decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, + matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + load: async scope => (await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId }) + .from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"), + scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined))) + .map(row => ({ ...row, dueAt: Date.now() })), + check: async entry => { + if (!(await owner.enabled())) return null; + 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 ${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 84c7f92a8a6..ec0f13b7ac1 100644 --- a/server/src/services/task-watchdogs.ts +++ b/server/src/services/task-watchdogs.ts @@ -1,4 +1,5 @@ import { notifyStatusCardIssueChange } from "./status-card-notifications.js"; +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; import { createHash } from "node:crypto"; import { and, asc, eq, gte, inArray, isNull, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -832,7 +833,7 @@ async function updateIssueWatchdogRow( return updated; } -export async function upsertIssueWatchdogForIssue( +async function upsertIssueWatchdogInTransaction( dbOrTx: any, companyId: string, issueId: string, @@ -848,12 +849,13 @@ export async function upsertIssueWatchdogForIssue( .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 }; } - const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx + const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx.transaction(async (tx: Db) => tx .insert(issueWatchdogs) .values({ companyId, @@ -871,7 +873,7 @@ export async function upsertIssueWatchdogForIssue( updatedAt: now, }) .returning() - .then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true })) + .then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true }))) .catch(async (error: unknown) => { if (!isIssueWatchdogUniqueConflict(error)) throw error; const winner = await dbOrTx @@ -886,6 +888,15 @@ export async function upsertIssueWatchdogForIssue( return { watchdog: toIssueWatchdog(insertResult.row), created: insertResult.created }; } +export function upsertIssueWatchdogForIssue( + dbOrTx: Db | Parameters[0]>[0], + companyId: string, + issueId: string, + input: IssueWatchdogUpsertInput, +): Promise<{ watchdog: IssueWatchdog; created: boolean }> { + return dbOrTx.transaction(tx => upsertIssueWatchdogInTransaction(tx, companyId, issueId, input)); +} + export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) { const issuesSvc = issueService(db); @@ -1482,7 +1493,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) return created; } - async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null } = {}) { + async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null; onDeferredDeadline?: (at: number) => void } = {}) { const watchdog = await markTerminalWatchdogIssueReviewed(row, opts); const sourceIssue = await db .select() @@ -1495,6 +1506,13 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) const input = await collectClassifierInput(watchdog.companyId, watchdog); const classification = classifyTaskWatchdogSubtree(input); + if (classification.state === "pending_first_run") { + const pending = new Set(classification.pendingIssueIds); + for (const issue of input.issues) if (pending.has(issue.id)) { + const createdAt = toEpochMs(issue.createdAt); + if (createdAt !== null) opts.onDeferredDeadline?.(createdAt + TASK_WATCHDOG_FIRST_RUN_GRACE_MS); + } + } if (classification.state !== "stopped") { return { state: classification.state, reason: classification.reason, classification }; } @@ -1614,13 +1632,14 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) }; } - async function listActiveWatchdogsForCompany(companyId?: string | null) { + async function listActiveWatchdogsForCompany(companyId?: string | null, watchdogId?: string) { return db .select() .from(issueWatchdogs) .where(and( eq(issueWatchdogs.status, "active"), ...(companyId ? [eq(issueWatchdogs.companyId, companyId)] : []), + watchdogId ? eq(issueWatchdogs.id, watchdogId) : undefined, )); } @@ -1750,14 +1769,16 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) issueId: string, actor: ActorFields = {}, ): Promise => { - await assertWatchedIssue(db, companyId, issueId); - const existing = await db + return db.transaction(async tx => { + await assertWatchedIssue(tx, companyId, issueId); + const existing = await tx .select() .from(issueWatchdogs) .where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId))) .then((rows) => rows[0] ?? null); if (!existing || existing.status === "disabled") return null; - const [updated] = await db + await notifyIssueRecovery(tx, { companyId }); + const [updated] = await tx .update(issueWatchdogs) .set({ status: "disabled", @@ -1769,14 +1790,17 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) .where(eq(issueWatchdogs.id, existing.id)) .returning(); return toIssueWatchdog(updated); + }); }, reconcileTaskWatchdogs: async (opts: { companyId?: string | null; runId?: string | null; + watchdogId?: string; + onDeferredDeadline?: (at: number) => void; issueCreatedAtGte?: Date | null; } = {}) => { - let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null); + let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null, opts.watchdogId); if (opts.issueCreatedAtGte) { const watchdogIssueIds = [...new Set(rows.map((row) => row.issueId))]; const eligibleIssueIds = new Set( @@ -1806,7 +1830,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) }; for (const row of rows) { result.checked += 1; - const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null }); + const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null, onDeferredDeadline: opts.onDeferredDeadline }); if (evaluated.state === "triggered") { result.triggered += 1; result.watchdogIssueIds.push(evaluated.watchdogIssueId);