diff --git a/doc/plans/2026-10-30-issue-recovery-sleep.md b/doc/plans/2026-10-30-issue-recovery-sleep.md index f191d5dca42..0c6cdf6b4d3 100644 --- a/doc/plans/2026-10-30-issue-recovery-sleep.md +++ b/doc/plans/2026-10-30-issue-recovery-sleep.md @@ -1,6 +1,6 @@ # Issue recovery without periodic scans -The issue recovery 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. +The issue recovery owners reconstruct their current obligations once at startup. Ordinary comment writes invalidate only the affected issue for assignment and lock recovery. Lifecycle, graph, run, configuration and eligibility changes invalidate 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 | | --- | --- | --- | --- | @@ -11,6 +11,10 @@ The issue recovery owners reconstruct their current obligations once at startup. | 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. | +Local comment hints do not broaden into an assignment check for every open task. A real PostgreSQL regression starts with1,000 open assigned issues, writes one comment through `issueService.addComment`, and observes exactly one assignment check; a subsequent company eligibility notification still checks all1,000. Restoring the previous comment hint makes that same regression check all1,000 after the comment. The test substitutes the expensive assignment decision only to count its invocations; it uses the real transaction, notification, coordinator and paged SQL. Separate producer acceptance cases exercise real recovery decisions and process execution. + +Parent/descendant disposition repair and shared execution references make status and run changes broader than a comment. Their existing notification boundaries deliberately retain company scope. Orphan blockers, active recovery actions, resolved dependencies and watchdogs also keep their company/subtree duties; an issue comment does not drop those independent obligations. The scope serializer selects known primitive fields and never serializes an entire insert or update object. + Reconstruction uses keyset pages of at most100rows. Repeated company hints coalesce reconstruction for30seconds. Already-known entry deadlines preempt that spacing without an early probe on each comment. Scheduling suppression is read once per pass. A failed read keeps scoped backoff debt. The two subsidiary company repairs do not run inside the per-issue loop. Legacy direct aggregate API calls retain both repairs. 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. diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts index b3d65f4dbcc..e23a672c0dc 100644 --- a/server/src/__tests__/issue-recovery-deadlines.test.ts +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -1,3 +1,4 @@ +import { setImmediate as immediate } from "node:timers/promises"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; @@ -8,6 +9,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { heartbeatService } from "../services/heartbeat.js"; import { issueService } from "../services/issues.js"; import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { registerIssueRecoveryWork } from "../services/recovery/work.js"; import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js"; import { taskWatchdogService } from "../services/task-watchdogs.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -27,6 +29,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea let draining = false; const tasks = new Map number | null; hasPending: () => Promise }>(); const errors: unknown[] = []; + const passes = new Set>(); const extraStops: Array<() => Promise> = []; const extraDeadlines: Array<() => number | null> = []; beforeAll(async () => { @@ -52,7 +55,14 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea 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); }; + coordinator.register = (queue, task) => { + tasks.set(queue, task); + return register(queue, { ...task, run: signal => { + const work = task.run(signal); passes.add(work); + void work.then(() => passes.delete(work), () => passes.delete(work)); + return work; + } }); + }; worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining); dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining); cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining); @@ -89,6 +99,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea } async function quiet() { await settled(); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute"); const selected = reads.mock.calls.length, executed = executes.mock.calls.length; if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); @@ -97,15 +108,32 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea } async function finish(issueId: string) { await issueService(db).update(issueId, { status: "done" }); - await heartbeat.drainActiveRunExecutions(); + // Closing one blocker can legitimately start its dependent. Wait for this + // task's dispatch and execution; quiet() verifies the whole fixture later. + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(and( + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + sql`${heartbeatRuns.status} in ('queued', 'running')`, + ))).toHaveLength(0); + }, { timeout: 15_000 }); const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); - expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); if (saved.length > 0) { const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); - await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1); + // Keep PostgreSQL I/O and coordinator settlement ahead of virtual time. + // A single leap can otherwise occur before the retry deadline is armed. + while (Date.now() <= due) { + await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1)); + await Promise.allSettled([...passes]); + await immediate(); + } await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(500); + await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); expect(remaining).toHaveLength(0); @@ -196,6 +224,52 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active"); expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet(); }); + it("limits an ordinary comment to its issue among one thousand open assignments", async () => { + const f = await fixture(); + const rows = await db.insert(issues).values(Array.from({ length: 1000 }, (_, index) => ({ + companyId: f.companyId, title: `Open work ${index}`, status: "todo", assigneeAgentId: f.agentId, + }))).returning({ id: issues.id }); + const checked: string[] = []; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const pass = task.run(signal); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); + return pass; + } }); + // Observe the actual paged reconstruction boundary without executing one + // thousand adapters. The producer, transaction, coordinator and SQL are real; + // the other acceptance cases exercise the resulting recovery decisions. + worker = registerIssueRecoveryWork(db, coordinator, { + stranded: async scope => { checked.push(scope.issueId); }, + orphanBlockers: async () => {}, recoveryActions: async () => {}, + dependencies: async () => ({ enqueueFailed: 0 }), locks: async () => {}, + watchdogs: async () => ({ incomplete: 0 }), schedulingSuppressed: async () => false, + }, () => true); + await worker.ready; + async function settleWork() { + for (let step = 0; step < 120; step++) { + await vi.advanceTimersByTimeAsync(1000); + await Promise.allSettled([...passes]); await immediate(); + if (coordinator.nextWakeAt() === null && idleWorkSnapshot().active === 0) return; + } + throw new Error("Recovery did not settle its committed work within two virtual minutes"); + } + await settleWork(); + expect(new Set(checked).size).toBe(1000); + checked.length = 0; + await issueService(db).addComment(rows[0].id, "New evidence on this task", { userId: "recovery-user" }); + await settleWork(); + expect(checked).toEqual([rows[0].id]); + // Company eligibility changes still reconstruct all assignments. Narrowing + // comment hints must not hide unrelated work after an operator resumes it. + checked.length = 0; + await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId })); + await settleWork(); + expect(new Set(checked).size).toBe(1000); + 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 }); diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts index cff04619945..32cd4407dd6 100644 --- a/server/src/services/issue-comment-writes.ts +++ b/server/src/services/issue-comment-writes.ts @@ -9,7 +9,8 @@ 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 notifyIssueRecovery(tx, { companyId: input.companyId, + issueId: typeof input.issueId === "string" ? input.issueId : undefined }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`); return tx.insert(issueComments).values(input).returning(); }); diff --git a/server/src/services/issue-recovery-notifications.ts b/server/src/services/issue-recovery-notifications.ts index 3c974228d86..7b709960831 100644 --- a/server/src/services/issue-recovery-notifications.ts +++ b/server/src/services/issue-recovery-notifications.ts @@ -1,11 +1,16 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export type IssueRecoveryScope = { companyId: string }; +/** An issue hint is reserved for local facts such as comments. Lifecycle, + * graph, run and eligibility changes retain company scope because their + * consequences can include ancestors and shared execution references. */ +export type IssueRecoveryScope = { companyId: string; issueId?: string }; export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) { - const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + const companyHint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + const issueHint = scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*"; for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) { - await notifyDeliveryWork(transaction, queue, hint); + await notifyDeliveryWork(transaction, queue, + queue === DELIVERY_QUEUES.strandedAssignments || queue === DELIVERY_QUEUES.staleIssueLocks ? issueHint : companyHint); } } diff --git a/server/src/services/native-status-reassessment-notifications.ts b/server/src/services/native-status-reassessment-notifications.ts index fe30a7b9bfa..3fb378c7605 100644 --- a/server/src/services/native-status-reassessment-notifications.ts +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -8,7 +8,7 @@ export type NativeStatusReassessmentScope = { companyId: string; issueId?: strin export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { await notifyStatusCardIssueChange(tx, scope.companyId); await notifyIssueAttentionRetentionWork(tx, { companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId }); - await notifyIssueRecovery(tx, scope); + await notifyIssueRecovery(tx, scope ? { companyId: scope.companyId } : undefined); 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/work.ts b/server/src/services/recovery/work.ts index 8bf7b75c7f4..407e4f1eb6d 100644 --- a/server/src/services/recovery/work.ts +++ b/server/src/services/recovery/work.ts @@ -32,13 +32,14 @@ export function registerIssueRecoveryWork( reconstructionIntervalMs: 30_000, beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, - matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + matches: (entry, scope) => !scope || (entry.companyId === scope.companyId && (!scope.issueId || entry.id === scope.issueId)), load: async scope => { - const key = scope?.companyId ?? "*"; + const key = scope ? JSON.stringify([scope.companyId, scope.issueId]) : "*"; if (!enabled) { cursors.delete(key); return []; } const cursor = cursors.get(key); const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and( scope ? eq(issues.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(issues.id, scope.issueId) : undefined, cursor ? gt(issues.id, cursor) : undefined, statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)), statuses ? isNull(issues.hiddenAt) : undefined, diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 9e8784b7481..cba58357cf0 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { - await notifyIssueRecovery(transaction, scope); + await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);