fix: scope comment-triggered recovery to the affected issue

Preserve company reconstruction for graph, run and eligibility changes. A real PostgreSQL producer regression verifies one assignment check among1000 open tasks, with broader eligibility reconstruction still intact.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

server/src/__tests__/issue-recovery-deadlines.test.ts

@@ -9,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";
@@ -223,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 });

server/src/services/issue-comment-writes.ts

@@ -7,7 +7,8 @@ type WriteDb = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
 
 export function insertIssueComment(db: WriteDb, input: PgInsertValue<typeof issueComments> & { companyId: string }) {
   return db.transaction(async tx => {
-    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();
   });

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);
   }
 }

server/src/services/native-status-reassessment-notifications.ts

@@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
 
 export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
-  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 }));
 }

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,

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);