Merge commit 'de500e03ec0f633070293c8e37ce226d143ebfa4' into codex/sleep-09-aggregate

* commit 'de500e03ec0f633070293c8e37ce226d143ebfa4':
  Remove trailing blank line in refresh notification module
  Remove unused aggregate watchdog diagnostics
  Bound recovery qualification database pools
  test: settle delegated chat completion before plan assertions
  Retain skipped-lock settlement and failed native cancellation work
  test(github): isolate scheduler time from postgres housekeeping
  test: settle saved retry cleanup before idle assertion
  fix: wake retry scheduling only when retry state changes
  fix(external-objects): scope issue deletion activation
  Update output progress test doubles for transactional writes
  Separate controller lease evidence from workspace ownership
  fix: retain sandbox attachment run scope for backup cleanup
  Preserve verified legacy process stop compatibility

doc/plans/2026-10-30-external-objects-sleep.md

@@ -59,7 +59,7 @@ and fences automatic result persistence. It does not modify that draft PR.
 | Issue title/description, comment and document references | Existing sync methods register object intent before mention replacement and stable identity intent before object upsert. Routes, runtime tools, adapters and plugin paths reuse these methods. |
 | Comment/document disappearance | Sync derives the retained mention owners and registers only those companies before deleting obsolete mentions. |
 | Last mention removal and reattachment | Mention replacement invalidates company deadlines. Retained objects stay available for reattachment. |
-| Issue deletion | Issue service registers its company before cascading mention deletion. |
+| Issue deletion | The issue service selects distinct mentioned object IDs by company and `sourceIssueId`, then registers only those IDs before the cascade. The source owner includes title, description, comments and documents. |
 | Company status/archive/deletion, including budget pause/resume | Company and budget services register company reconstruction before status or deletion writes. |
 | Feature enable/disable | Instance settings registers global reconstruction before the experimental-settings write. |
 | Plugin manifest/status/uninstall | Plugin registry registers reconstruction before lifecycle writes. Installed plugins acquire support through the ordinary ready transition. |

doc/plans/2026-10-30-live-recovery-sleep.md

@@ -5,12 +5,12 @@ This is the final recovery slice after run dispatch, issue state recovery, and d
 | Duty | Durable truth and pending predicate | Producers | Deadline and exit |
 | --- | --- | --- | --- |
 | Resultless native session | Native running/failed run; finalization `retryable_failure`; no result | Execution/finalization writes, checkpoint writes, restart claim writes, process identity events, and existing run-dispatch/deferred-eligibility boundaries for agent, task, review, dependency and pause-hold changes | Maximum saved retry/claim lease deadline. Fresh claim owner validates provider/session state. Accepted result, terminal failure, or explicit ownership gate removes the entry. |
-| Native question cancellation | Cancellable run with committed `nativeQuestionCancellation` marker | Authorized task-completion interaction transaction | Immediate. Existing cancellation owner consumes exact company/run marker. Terminal status removes debt. |
-| Active execution | Running run | Run lifecycle, controller renewal/revocation, native authority and process evidence writes | Fresh native/legacy controller expiry, saved five-minute preparation grace, or bounded physical-process probe. Local execution promises retain execution ownership. Explicit evidence/manual gates park until their owning producer changes truth. |
-| Settled native process identity | Terminal native failure/cancellation with persisted process identity | Terminal execution writes, server process identity/stop receipts, environment settlement | Immediate inspection. Only matched local host/PID/group/birth or confirmed remote stop evidence clears identity. A live matched process or local execution owner keeps a bounded probe. Missing authority waits for committed evidence. Startup reconstructs rows independently of the old capped process Map. Legacy local PID rows lacking birth evidence remain unchanged and excluded until the actual process-metadata writer records matching identity; they grant no automatic continuation authority. Remote allocations remain eligible without local birth evidence because their exact provider stop receipts own release. Successful release removes any obsolete compatibility Map entry. |
+| Native question cancellation | Cancellable run or resultless failed native retry with committed `nativeQuestionCancellation` marker | Authorized task-completion interaction transaction | Immediate. Existing cancellation owner consumes exact company/run marker. Failed retries remain pending through retryable failure or an exactly bound pending/acknowledged native retry Stop receipt. Locked dispatch and exact acknowledged cancellation CAS retain authority. Completed terminal history removes debt. |
+| Active execution | Running run | Run lifecycle, controller renewal/revocation, native authority and process evidence writes | Fresh native/legacy controller expiry, saved five-minute preparation grace, or bounded physical-process probe. Local execution promises retain execution ownership. A native controller deadline requires saved boot, PID and birth identity; a workspace-finalizer lease expiry alone grants no execution authority. Explicit evidence/manual gates park until their owning producer changes truth. |
+| Settled native process identity | Terminal native failure/cancellation with persisted process identity | Terminal execution writes, server process identity/stop receipts, environment settlement | Immediate inspection. Only matched local host/PID/group/birth or confirmed remote stop evidence clears identity. A live matched process or local execution owner keeps a bounded probe. Missing authority waits for committed evidence. Startup reconstructs rows independently of the old capped process Map. Legacy rows without server-written local identity remain unchanged and excluded until actual process metadata arrives. A missing stored birth alone preserves prior compatibility only with exact server-written host/PID/group identity, a valid PID equal to its group, and positive absence of both; the stop receipt remains available to existing continuation validation. Known birth times still require equality. Remote allocations remain eligible without local birth evidence because their exact provider stop receipts own release. Successful release removes any obsolete compatibility Map entry. |
 | Orphan active environment lease | Active lease with NULL run reference or terminal same-company run | Environment owning transactions and indexed physical-resource peer notifications | Saved lease updatedAt plus five minutes, re-read before due work. Existing export, shutdown, and shared-resource fences remain. Workspace preservation lock/statement timeouts retain a bounded thirty-second retry; claim/evidence gates still park. Exact-lease dispatch does not mutate timestamps for broad-page fairness. Successful CAS hands work to the existing pending-cleanup owner. |
 | Legacy shutdown settlement | Exact lease with pending shutdown marker | Begin, renew, finish, expiry settlement owning transactions | Exact saved marker deadline. Existing owner revalidates boot/controller, source, competing allocations and stop/copyback authority. Expiry records preserved workspace recovery; it never infers file export from process loss. |
-| Stranded agent status | Ready agent with status running | Agent configuration/lifecycle/instruction settlement, run/resource transitions, accounting acknowledgement, canonical wake status writes, and direct existing-receipt terminal writers | Saved updatedAt plus five minutes, then actual surviving controller expiry when present. Fresh settlement rechecks live execution, native controller, working copy, and lease truth. Settled status removes debt. |
+| Stranded agent status | Ready agent with status running | Agent configuration/lifecycle/instruction settlement, run/resource transitions, accounting acknowledgement, canonical wake status writes, and direct existing-receipt terminal writers | Saved updatedAt plus five minutes, then actual surviving controller expiry when present. Fresh settlement rechecks live execution, native controller, working copy, and lease truth. A genuine skipped agent row lock retains a thirty-second retry after confirming the row still exists; ordinary eligibility refusals park. Settled status removes debt. |
 | Active output silence | Running run and coalesced last output/process start/run start/created timestamp | Actual output progress transaction, terminal source mutation, same-run terminal activity commit, watchdog decisions, execution state | Five-minute threshold, exact saved snooze/continue expiry, or physical termination probe. Fresh check re-reads output and decision. Existing watchdog alone owns completion evidence, process cleanup and fold. Suppressed/manual/blocked history does not poll. |
 
 All reconstruction queries use company/run/lease/agent scopes and keyset pages. Run and resource topics discard irrelevant fields before coalescing. The generic scheduler's pending/deadline reads use memory. Ordinary writes register intent before mutation in the same owning transaction. Nested commit, outer rollback and uncertain transaction reconciliation retain the shared database-work semantics.
@@ -49,3 +49,9 @@ The regression holds the real task row `FOR UPDATE` until the existing one-secon
 A successful legacy lease renewal signals only its active-execution run deadline. The private liveness pool reuses its already-owned transaction for statement timeout, intent, and update; the public renewal API still owns an atomic transaction. Other controller/run transitions retain their broader producer signals. Renewal still uses `clock_timestamp()` and cannot resurrect an expired lease.
 
 The actual private-pool control prewarms the connection and requires exactly five transaction statements, one XID, one active-execution topic, no savepoints, and no other recovery topics. The registered-owner control measures the bounded intent/settlement passes, proves no unrelated queue pass, crosses the prior deadline, and requires the fresh saved expiry. The existing controller suite covers real timeout retries, foreign/current boots, expired refusal, revocation, and shared-pool saturation. Dead aggregate guard branches remain for the final aggregate cutover PR to remove together.
+
+## Cancellation and settlement races
+
+Agent settlement retains a bounded retry only after its actual `FOR UPDATE SKIP LOCKED` attempt misses an existing candidate. Releasing a transaction that only held the lock emits no eligibility hint; the retained entry still settles exactly once and becomes quiet.
+
+A native question marker can commit before execution changes the run to failed/retryable failure. Both reconstruction and consumption select this exact unfinished class. A Stop may also advance the coordinator to `terminal_failure/native_retry_cancelled` before the final run-status commit fails. The owning pending predicate includes only the existing exactly bound, audited pending/acknowledged cancellation receipt; cancellation candidate discovery uses `nativeRetryCancellationEligible` again. Existing locked native dispatch and the unchanged acknowledged-receipt final CAS still authorize all effects. A one-shot PostgreSQL failure proves this continuation completes with the same intent/audit receipt and no provider execution, while marked failed history with misbound receipts remains unchanged and quiet.

doc/plans/2026-10-30-run-dispatch-sleep.md

@@ -12,6 +12,15 @@ invalidate retries notify the retry worker in their transaction. Promotion uses
 the existing gate and compare-and-set, and its committed write wakes ordinary
 queued dispatch. A queued run behind occupied capacity waits for the owning
 run's lifecycle transition. Configuration and authority changes also wake it.
+Retry-schedule notifications are separate from ordinary queue eligibility.
+`notifyRunDispatch` wakes only queued dispatch; ordinary issue, agent, budget
+and company edits cannot repeatedly scan the global retry schedule.
+`notifyScheduledRetryDispatch` wakes retry reconciliation as well as dispatch
+for actual retry creation, rescheduling and cancellation. Its producers include
+bounded and operator-requested retries, recovery retries, native capacity and
+safe-replacement successors, and the existing retry-cancellation transition
+boundary. Shared runtime changes do not impose a spacing floor on a saved deadline.
+
 An idle queue has no recurring read or timer-driven dispatch attempt. Notifications batch for one second and carry the changed company to queued dispatch. Known retry deadlines still preempt batching. Cancellation-owned retries stay with their cancellation worker and do not retain a promotion deadline.
 
 The supported writer contract is service writes through the shared database

server/src/__tests__/chat-channels.integration.test.ts

@@ -76890,7 +76890,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
           resolveSessionBeforeForWakeup: async () => null,
         });
         return adapter.withIssueExecutionLock({ companyId: context.fixture.companyId, runId: releaseRunId, now: new Date() }, async (_locked, ports) => {
-          const target = { companyId: context.fixture.companyId, wakeId: context.action.id, now: new Date() };
+          const target = { companyId: context.fixture.companyId, wakeId: context.action.id, agentId: context.fixture.assignedAgentId, now: new Date() };
           if (change === "queue rejection") await ports.transaction.cancelDeferredWake({ ...target, reason: "PRIVATE rejection" });
           else await ports.transaction.failDeferredWake(target);
           return { outcome: { kind: "released" }, postCommitEffects: [] };

server/src/__tests__/deferred-execution-receipt-work.test.ts

@@ -78,7 +78,15 @@ describe("deferred receipt durable dispatch", () => {
     return { companyId, agent, issue, source, actorId };
   }
   async function quiet() {
-    await vi.waitFor(async () => { expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }, { timeout: 40_000 });
+    await vi.waitFor(async () => {
+      if (vi.isFakeTimers()) {
+        // Fake-clock cases have no live execution. Settle finite cleanup
+        // deadlines explicitly instead of advancing at waitFor's wall speed.
+        await vi.advanceTimersByTimeAsync(1000);
+        await Promise.allSettled([...passes]); await immediate();
+      }
+      expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
+    }, { timeout: 40_000 });
     const reads = vi.spyOn(db, "select"), 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"] });
@@ -455,8 +463,8 @@ describe("deferred receipt durable dispatch", () => {
       const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!));
       expect(wake.runId, JSON.stringify(wake)).toBeTruthy(); runId = wake.runId;
     }, { timeout: 40_000 });
+    // Observe this Retry before closing the task and draining its follow-ups.
     await vi.waitFor(async () => {
-      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 })).toBe("succeeded");
       expect(run.stdoutExcerpt).toContain("receipt-proof");

server/src/__tests__/external-object-refresh-work.test.ts

@@ -2,11 +2,12 @@ import { createServer } from "node:http";
 import { randomUUID } from "node:crypto";
 import { setTimeout as delay } from "node:timers/promises";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
-import { eq, sql } from "drizzle-orm";
+import { and, eq, inArray, sql } from "drizzle-orm";
 import { activityLog, approvals, applyPendingMigrations, databaseWorkPending, budgetIncidents, budgetPolicies, companies, costEvents, createDb, externalObjects, externalObjectMentions, issues, issueComments, plugins } from "@paperclipai/db";
 import { externalObjectService, type ExternalObjectResolveResult } from "../services/external-objects.js";
 import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
 import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
+import { issueService } from "../services/issues.js";
 import { documentService } from "../services/documents.js";
 import { budgetService } from "../services/budgets.js";
 import { pluginRegistryService } from "../services/plugin-registry.js";
@@ -367,6 +368,94 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: {
     expect(provider).toHaveBeenCalledTimes(2);
   });
 
+  it("scopes repeated issue deletions to cascaded source objects and cancels their last future mention", async () => {
+    clock(); const svc = service();
+    const own = await fixture(), foreign = await fixture();
+    const future = new Date(Date.now() + 86_400_000), removedDeadline = new Date(Date.now() + 1_800_000);
+    const sharedUrl = "https://example.com/shared-issue-object";
+    const [retained] = await db.insert(issues).values({ companyId: own.companyId, title: "Retained",
+      description: sharedUrl }).returning();
+    await svc.syncIssue(retained!.id);
+    await db.update(issues).set({ description: sharedUrl }).where(eq(issues.id, foreign.issueId));
+    await svc.syncIssue(foreign.issueId);
+    const deletions: Array<{ id: string; affected: string[]; orphaned: string[] }> = [];
+    for (let index = 0; index < 3; index++) {
+      const [issue] = await db.insert(issues).values({ companyId: own.companyId,
+        title: sharedUrl, description: `https://example.com/deleted-${index}` }).returning();
+      await svc.syncIssue(issue!.id);
+      const [comment] = await db.insert(issueComments).values({ companyId: own.companyId, issueId: issue!.id,
+        body: `https://example.com/comment-${index}` }).returning();
+      await svc.syncComment(comment!.id);
+      const doc = await documentService(db).upsertIssueDocument({ issueId: issue!.id, key: "notes", format: "markdown",
+        body: `https://example.com/document-${index}` });
+      await svc.syncDocument(doc.document.id);
+      const mentions = await db.select().from(externalObjectMentions).where(and(
+        eq(externalObjectMentions.companyId, own.companyId), eq(externalObjectMentions.sourceIssueId, issue!.id)));
+      expect(new Set(mentions.map(mention => mention.sourceKind))).toEqual(new Set(["title", "description", "comment", "document"]));
+      deletions.push({ id: issue!.id, affected: [...new Set(mentions.map(mention => mention.objectId!))],
+        orphaned: mentions.filter(mention => mention.sourceKind !== "title").map(mention => mention.objectId!) });
+    }
+    const shared = (await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.sourceIssueId, retained!.id)))[0]!.objectId!;
+    for (let offset = 0; offset < 1100; offset += 250) {
+      const ids = Array.from({ length: Math.min(250, 1100 - offset) }, () => randomUUID());
+      await db.insert(externalObjects).values(ids.map(id => ({ id, companyId: own.companyId,
+        providerKey: "url", objectType: "link", externalId: id, nextRefreshAt: future })));
+      await db.insert(externalObjectMentions).values(ids.map(objectId => ({ companyId: own.companyId,
+        sourceIssueId: own.issueId, sourceKind: "description" as const, objectId })));
+    }
+    await db.update(externalObjects).set({ nextRefreshAt: future });
+    for (const entry of deletions) for (const id of entry.orphaned)
+      await db.update(externalObjects).set({ nextRefreshAt: removedDeadline }).where(eq(externalObjects.id, id));
+    await start(svc);
+    await settle(() => coordinator!.nextWakeAt() === removedDeadline.getTime() && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+    expect(provider).not.toHaveBeenCalled();
+    const reads: Array<{ query: string; params: unknown[] }> = [];
+    const selections: Array<{ query: string; params: unknown[] }> = [];
+    const debug = db.$client.options.debug;
+    db.$client.options.debug = (_connection, query, params) => {
+      if (/^select/i.test(query) && query.includes('from "external_objects"') && query.includes('inner join "companies"'))
+        reads.push({ query, params: [...params] });
+      if (/^select distinct/i.test(query) && query.includes('from "external_object_mentions"'))
+        selections.push({ query, params: [...params] });
+    };
+    try {
+      for (let index = 0; index < deletions.length; index++) {
+        const entry = deletions[index], first = reads.length;
+        const result = await issueService(db).remove(entry.id);
+        expect(result?.id).toBe(entry.id);
+        const next = index === deletions.length - 1 ? future : removedDeadline;
+        await settle(() => reads.length > first && coordinator!.nextWakeAt() === next.getTime()
+          && !databaseWorkPending(db, DELIVERY_QUEUES.externalObjectRefresh));
+        const deletionReads = reads.slice(first);
+        expect(deletionReads.length).toBeLessThanOrEqual(entry.affected.length);
+        for (const read of deletionReads) {
+          expect(read.query).toContain('"external_objects"."id" =');
+          expect(read.params).toContain(own.companyId); expect(read.params).not.toContain(foreign.companyId);
+          expect(read.params.some(value => entry.affected.includes(String(value)))).toBe(true);
+        }
+      }
+      expect(selections).toHaveLength(deletions.length);
+      for (let index = 0; index < selections.length; index++) {
+        expect(selections[index]!.query).toContain('"external_object_mentions"."company_id" =');
+        expect(selections[index]!.query).toContain('"external_object_mentions"."source_issue_id" =');
+        expect(selections[index]!.params).toContain(own.companyId); expect(selections[index]!.params).toContain(deletions[index]!.id);
+      }
+      const lastMentions = await db.select().from(externalObjectMentions).where(eq(externalObjectMentions.objectId, shared));
+      expect(lastMentions).toHaveLength(1); expect(lastMentions[0]!.sourceIssueId).toBe(retained!.id);
+      const orphanIds = deletions.flatMap(entry => entry.orphaned);
+      const orphanRows = await db.select().from(externalObjects).where(inArray(externalObjects.id, orphanIds));
+      expect(orphanRows).toHaveLength(orphanIds.length);
+      for (const orphan of orphanRows) expect(orphan.nextRefreshAt).toEqual(removedDeadline);
+      expect(await db.select().from(externalObjectMentions).where(inArray(externalObjectMentions.objectId, orphanIds))).toEqual([]);
+      const foreignRows = await db.select().from(externalObjects).where(eq(externalObjects.companyId, foreign.companyId));
+      expect(foreignRows).toHaveLength(1); expect(foreignRows[0]!.nextRefreshAt).toEqual(future);
+      const count = reads.length;
+      await vi.advanceTimersByTimeAsync(removedDeadline.getTime() - Date.now() + 1);
+      expect(reads).toHaveLength(count); expect(provider).not.toHaveBeenCalled();
+      expect(coordinator!.nextWakeAt()).toBe(future.getTime());
+    } finally { db.$client.options.debug = debug; }
+  });
+
   it("keeps warm standby startup SQL-free and resumes a first committed producer", async () => {
     clock(); admitted = false; reconcile = false;
     const svc = service(); const selects = vi.spyOn(db, "select");

server/src/__tests__/github-connection-work.test.ts

@@ -44,7 +44,11 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
   beforeAll(async () => {
     if (databaseUrl) await applyPendingMigrations(databaseUrl);
     else temporaryDatabase = await startEmbeddedPostgresTestDatabase("paperclip-github-work-");
-    db = createDb(databaseUrl ?? temporaryDatabase!.connectionString, { maxConnections: 4 });
+    // Only scheduler time is virtual here. Driver housekeeping must not expire
+    // a real socket connection while that fake clock advances across idle windows.
+    db = createDb(databaseUrl ?? temporaryDatabase!.connectionString, {
+      maxConnections: 4, idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0,
+    });
     // This explicitly assigned external test database is exclusive to this suite.
     await db.transaction(async tx => { await tx.execute(sql`SET LOCAL client_min_messages = warning`); await tx.execute(sql`TRUNCATE companies CASCADE`); });
     server = createServer(async (req, res) => {
@@ -409,7 +413,8 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
     const schedule = f.service.githubConnectionContinuitySchedule.bind(f.service);
     const load = vi.spyOn(f.service, "githubConnectionContinuitySchedule").mockImplementation(companyId => { loadTimes.push(Date.now()); return schedule(companyId); });
     const sweep = vi.spyOn(f.service, "sweepGitHubConnectionContinuity");
-    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true });
+    const errors: unknown[] = [];
+    const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
     const continuity = registerGitHubConnectionContinuityWork(db, coordinator, f.service, () => true);
     const events = registerGitHubConnectionEventWork(db, coordinator, githubConnectionEventService(db, { connector: f.connector }), () => true);
     try {
@@ -423,6 +428,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
       }
       await vi.advanceTimersByTimeAsync(10_000);
       expect(load).not.toHaveBeenCalled(); expect(sweep).not.toHaveBeenCalled(); expect(leases).toBe(1);
+      expect(errors).toEqual([]);
       for (let index = 0; index < 30; index++) await db.transaction(async tx => { await notifyGitHubCredentialChange(tx, f.connection); });
       expect(leases).toBeLessThanOrEqual(2);
       await vi.advanceTimersByTimeAsync(6_000);
@@ -442,6 +448,7 @@ describeDatabase("GitHub coordinator with PostgreSQL and controlled Cloud HTTP (
       const quiet = { leases, loads: load.mock.calls.length };
       await vi.advanceTimersByTimeAsync(60_000);
       expect({ leases, loads: load.mock.calls.length }).toEqual(quiet);
+      expect(errors).toEqual([]);
     } finally { await events.stop(); await continuity.stop(); await coordinator.stop(); }
   });
 

server/src/__tests__/heartbeat-run-dispatch-lifecycle.test.ts

@@ -3,7 +3,7 @@ import { PgDialect } from "drizzle-orm/pg-core";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
-import { notifyRunDispatch } from "../services/run-dispatch-notifications.js";
+import { notifyRunDispatch, notifyScheduledRetryDispatch } from "../services/run-dispatch-notifications.js";
 import { registerRunDispatchWork } from "../services/heartbeat/dispatch-work.js";
 
 const cleanup: Array<() => Promise<void>> = [];
@@ -38,6 +38,7 @@ function fixture(input: { due?: number; suppressed?: boolean; draining?: boolean
   }, () => !state.draining && !state.standby);
   cleanup.push(() => coordinator.stop());
   return { db, state, promote, next, dispatch, coordinator, worker,
+    notifyRetry: () => db.transaction(tx => notifyScheduledRetryDispatch(tx)),
     notify: (scope?: { companyId: string }) => db.transaction(tx => notifyRunDispatch(tx, scope)),
     settle: () => vi.advanceTimersByTimeAsync(1_000) };
 }
@@ -61,9 +62,9 @@ describe("run dispatch obligations", () => {
   });
   it("rearms earlier work and cancels a future deadline on a committed producer change", async () => {
     const f = fixture({ due: 1_600_000 }); await f.worker.ready;
-    f.state.due = 1_005_000; await f.notify(); await f.settle();
+    f.state.due = 1_005_000; await f.notifyRetry(); await f.settle();
     expect(f.coordinator.nextWakeAt()).toBe(1_005_000);
-    f.state.due = null; await f.notify(); await f.settle();
+    f.state.due = null; await f.notifyRetry(); await f.settle();
     expect(f.coordinator.nextWakeAt()).toBeNull();
     const before = f.next.mock.calls.length;
     await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.next).toHaveBeenCalledTimes(before);
@@ -76,7 +77,7 @@ describe("run dispatch obligations", () => {
       await f.notify(source);
       await vi.advanceTimersByTimeAsync(10);
     }
-    expect(f.promote.mock.calls.length).toBeLessThanOrEqual(2);
+    expect(f.promote).toHaveBeenCalledOnce();
     expect(f.dispatch.mock.calls.length).toBeLessThanOrEqual(2);
     await f.settle();
     expect(f.dispatch).toHaveBeenLastCalledWith({ companyId: "company-a" });
@@ -87,7 +88,7 @@ describe("run dispatch obligations", () => {
   });
   it("remembers the first producer arriving during drain without owner SQL", async () => {
     const f = fixture(); await f.worker.ready; f.state.draining = true;
-    await f.notify(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
+    await f.notifyRetry(); await f.settle(); await vi.advanceTimersByTimeAsync(90_000);
     expect(f.promote).toHaveBeenCalledOnce(); expect(f.next).toHaveBeenCalledOnce();
     expect(idleWorkSnapshot().active).toBe(0);
     f.state.draining = false; await vi.advanceTimersByTimeAsync(30_000);
@@ -96,7 +97,7 @@ describe("run dispatch obligations", () => {
   });
   it("releases a failed producer hold during drain and retains recovery for resume", async () => {
     const f = fixture(); await f.worker.ready;
-    f.state.fail = true; await f.notify(); await f.settle();
+    f.state.fail = true; await f.notifyRetry(); await f.settle();
     expect(f.promote).toHaveBeenCalledTimes(2);
     f.state.draining = true; await vi.advanceTimersByTimeAsync(90_000);
     expect(f.promote).toHaveBeenCalledTimes(2); expect(f.next).toHaveBeenCalledOnce();
@@ -109,8 +110,8 @@ describe("run dispatch obligations", () => {
     let release!: () => void;
     const held = new Promise<void>(resolve => { release = resolve; });
     f.promote.mockImplementationOnce(() => held);
-    await f.notify(); await f.settle();
-    await f.notify(); await f.settle();
+    await f.notifyRetry(); await f.settle();
+    await f.notifyRetry(); await f.settle();
     expect(f.promote).toHaveBeenCalledTimes(2);
     release(); await f.settle();
     await vi.advanceTimersByTimeAsync(30_000);
@@ -121,7 +122,7 @@ describe("run dispatch obligations", () => {
     await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).not.toHaveBeenCalled();
     f.state.standby = false; await vi.advanceTimersByTimeAsync(30_000);
     expect(f.promote).toHaveBeenCalledOnce(); expect(f.state.due).toBeNull();
-    f.state.due = Date.now() + 60_000; await f.notify(); await f.settle(); await f.worker.stop();
+    f.state.due = Date.now() + 60_000; await f.notifyRetry(); await f.settle(); await f.worker.stop();
     await vi.advanceTimersByTimeAsync(90_000); expect(f.promote).toHaveBeenCalledTimes(2);
   });
 });

server/src/__tests__/heartbeat-run-dispatch-work.test.ts

@@ -2,9 +2,11 @@ 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, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns } from "@paperclipai/db";
+import { agents, companies, costEvents, createDb, databaseWorkPending, heartbeatRuns, issues } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { issueService } from "../services/issues.js";
+import { materializeNativeProviderCapacityRetry } from "../services/native-runtime/native-provider-capacity-retry.js";
 import { agentService } from "../services/agents.js";
 import { budgetService } from "../services/budgets.js";
 import { heartbeatService } from "../services/heartbeat.js";
@@ -24,9 +26,14 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
   let reads: ReturnType<typeof vi.spyOn>;
   let draining = false;
   const errors: unknown[] = [];
+  const retryQueries: string[] = [];
   beforeAll(async () => {
     if (!external) database = await startEmbeddedPostgresTestDatabase("run-dispatch-work-");
     db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+    db.$client.options.debug = (_connection, query) => {
+      if (query.startsWith("select") && query.includes('"heartbeat_runs"') &&
+        query.includes("startupCancellation") && query.includes('"scheduled_retry_at"')) retryQueries.push(query);
+    };
     await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
   }, 90_000);
   beforeEach(async () => {
@@ -88,6 +95,45 @@ describe.skipIf(!support.supported)("run dispatch production wiring (controlled
     await fixture(); await worker!.stop(); await coordinator!.stop();
     await start(); await quiet();
   });
+  it.each(["agent", "issue"] as const)("does not scan global scheduled retries for repeated ordinary %s edits", async kind => {
+    const f = await fixture();
+    const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Edit source", status: "backlog" }).returning();
+    await start(); await quiet();
+    const before = retryQueries.length;
+    expect(before).toBeGreaterThan(0); // The passive observer saw real startup retry SQL.
+    for (let index = 0; index < 12; index++) {
+      if (kind === "agent") await agentService(db).update(f.agentId, { name: `Edited agent ${index}` });
+      else await issueService(db).update(issue.id, { title: `Edited issue ${index}` });
+      await new Promise(resolve => setTimeout(resolve, 100));
+    }
+    await quiet();
+    expect(retryQueries).toHaveLength(before);
+  });
+  it.each([false, true])("observes a native capacity retry only after its transaction settles (rollback=%s)", async rollback => {
+    const f = await fixture();
+    const [issue] = await db.insert(issues).values({ companyId: f.companyId, title: "Capacity retry", status: "in_progress",
+      assigneeAgentId: f.agentId }).returning();
+    await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.runId));
+    await start(); await quiet();
+    let retryId = "";
+    const writing = db.transaction(async tx => {
+      retryId = await materializeNativeProviderCapacityRetry({ tx: tx as unknown as typeof db, ...f, issueId: issue.id });
+      expect(databaseWorkPending(db, DELIVERY_QUEUES.scheduledRetryDispatch)).toBe(true);
+      expect(await status(retryId)).toBeUndefined();
+      if (rollback) throw new Error("native capacity rollback");
+    });
+    if (rollback) {
+      await expect(writing).rejects.toThrow("native capacity rollback");
+      await quiet(); expect(await status(retryId)).toBeUndefined();
+    } else {
+      await writing;
+      const retry = await status(retryId);
+      expect(retry.status).toBe("scheduled_retry");
+      await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(retry.scheduledRetryAt!.getTime()), { timeout: 5_000 });
+      await heartbeat.cancelRun(retryId, "fixture teardown", { suppressImmediateRecovery: true });
+      await quiet();
+    }
+  });
   it("ordinary retry creation wakes the worker, respects the exact deadline, dispatches and settles", async () => {
     const f = await fixture(); await start(); await quiet();
     const result = await heartbeat.scheduleBoundedRetry(f.runId, { delayMs: 4_000, random: () => 0 });

server/src/__tests__/live-recovery-work.test.ts

@@ -29,6 +29,9 @@ import { createAgentLifecycle, configureAgentLifecycle } from "../services/agent
 import { issueTreeControlService } from "../services/issue-tree-control.js";
 import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
 import { issueService } from "../services/issues.js";
+import { nativeRetryCancellationEligible, pendingFailedNativeRetryCancellationCondition } from "../services/native-runtime/native-cancellation-request.js";
+import { requestNativeQuestionRunCancellation } from "../services/native-runtime/native-question-bridge.js";
+import { readNativeLocalProcessStop } from "../services/native-local-process-stop.js";
 import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
 import { logger } from "../middleware/logger.js";
 
@@ -108,7 +111,7 @@ describe("residual recovery production matrix", () => {
     expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
     const quietAt = Date.now();
     if (vi.isFakeTimers()) await advance(31_000);
-    else await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 35_000 });
+    else { await realDelay(31_000); await Promise.allSettled([...passes]); }
     console.info("quiet-drain", JSON.stringify({ quietAt, now: Date.now(), active: idleWorkSnapshot().active, next: [...deadlines].map(([q, at]) => [q, at()]), passes: Object.fromEntries([...deadlines.keys()].map(queue => [queue, timeline.filter(event => Array.isArray(event) && event[0] === "pass" && event[1] === queue).length])), settled: timeline.filter(event => Array.isArray(event) && event[0] === "settled") }));
     await vi.waitFor(() => { expect(coordinator.nextWakeAt(), JSON.stringify([...deadlines].map(([name, at]) => [name, at()]))).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
     const queries: string[] = [], prior = db.$client.options.debug;
@@ -160,6 +163,78 @@ describe("residual recovery production matrix", () => {
     expect((await getRun(run.id)).resultJson).toMatchObject({ cancelledByIssueStatus: "done", cancelledIssueId: f.issue.id });
     await quiet();
   });
+  it.each([false, true])("finishes a question cancellation whose run fails into a pending native retry before consumption (partial Stop fault: %s)", async partialStopFault => {
+    const f = await fixture();
+    let stopReceipt: Record<string, unknown> | undefined;
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "running", runtimeMode: "native", nativeIssueId: f.issue.id, nativePhase: "awaiting_input",
+      contextSnapshot: { issueId: f.issue.id } }).returning();
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id,
+      runId: run.id, phase: "observed", recoveryState: "awaiting_evidence" });
+    if (partialStopFault) {
+      await db.execute(sql`create sequence question_cancel_failure_once`);
+      await db.execute(sql.raw(`create function question_cancel_failure_once() returns trigger language plpgsql as $$
+        begin if new.id = '${run.id}'::uuid and new.status = 'cancelled' and nextval('question_cancel_failure_once') = 1
+          then raise exception 'fixture question cancellation commit unavailable'; end if; return new; end $$`));
+      await db.execute(sql`create trigger question_cancel_failure_once before update on heartbeat_runs for each row execute function question_cancel_failure_once()`);
+      extraStops.push(async () => {
+        await db.execute(sql`drop trigger if exists question_cancel_failure_once on heartbeat_runs`);
+        await db.execute(sql`drop function if exists question_cancel_failure_once()`);
+        await db.execute(sql`drop sequence if exists question_cancel_failure_once`);
+      });
+    }
+    const marker = { version: 1, issueId: f.issue.id, kind: "issue_terminal", issueStatus: "done" };
+    const history = await db.insert(heartbeatRuns).values(Array.from({ length: 25 }, () => ({
+      companyId: f.companyId, agentId: f.agent.id, runtimeMode: "native", nativeIssueId: f.issue.id,
+      status: "failed", finishedAt: new Date(), nativePhase: "terminal_failure",
+      contextSnapshot: { nativeQuestionCancellation: marker }, resultJson: { nativeCancellation: {
+        schema: "paperclip.native-cancellation.v1", runId: randomUUID(), companyId: f.companyId, issueId: f.issue.id,
+        scope: "run", intentId: randomUUID(), intentAuditId: randomUUID(), dispatched: false, dispatchState: "pending",
+      } },
+    }))).returning();
+    await db.insert(nativeRunFinalizations).values(history.map(row => ({ companyId: f.companyId, issueId: f.issue.id,
+      runId: row.id, phase: "terminal_failure", failureCode: "native_retry_cancelled", recoveryState: "blocked" })));
+    draining = true; await start();
+    expect(await db.transaction(tx => requestNativeQuestionRunCancellation(tx, { companyId: f.companyId,
+      issueId: f.issue.id, sourceRunId: run.id, idempotencyKey: `paperclip-runner-question:${run.id}:raced-question`,
+      payload: { questionSet: {}, runtimeRequestId: "raced-question" },
+    }, { kind: "issue_terminal", issueStatus: "done" }))).toBe(run.id);
+    expect((await getRun(run.id)).contextSnapshot?.nativeQuestionCancellation).toMatchObject(marker);
+    await db.transaction(async tx => {
+      await notifyExecutionRunWork(tx, { companyId: f.companyId, runId: run.id, agentId: f.agent.id });
+      await tx.update(heartbeatRuns).set({ status: "failed", nativePhase: "retryable_failure", finishedAt: new Date() })
+        .where(eq(heartbeatRuns.id, run.id));
+      await tx.update(nativeRunFinalizations).set({ phase: "retryable_failure", recoveryState: "resuming_session",
+        nextAttemptAt: new Date(), leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, run.id));
+    });
+    draining = false; await advance(1_000);
+    if (partialStopFault) {
+      await vi.waitFor(() => expect(errors).toHaveLength(1));
+      expect(errors.shift()).toBeInstanceOf(AggregateError);
+      expect((await getRun(run.id)).status).toBe("failed");
+      const saved = await getRun(run.id);
+      stopReceipt = saved.resultJson?.nativeCancellation as Record<string, unknown>;
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
+      expect(coordinator).toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", resultId: null });
+      expect(nativeRetryCancellationEligible({ runId: run.id, companyId: f.companyId, issueId: f.issue.id,
+        resultJson: saved.resultJson }, coordinator)).toBe(true);
+      const [debt] = await db.select({ pending: pendingFailedNativeRetryCancellationCondition() }).from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, run.id));
+      expect(debt.pending).toBe(true);
+    }
+    await advance(30_000);
+    await vi.waitFor(async () => expect((await getRun(run.id)).status).toBe("cancelled"));
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0])
+      .toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null });
+    if (stopReceipt) expect((await getRun(run.id)).resultJson?.nativeCancellation).toMatchObject({
+      intentId: stopReceipt.intentId, intentAuditId: stopReceipt.intentAuditId,
+      acknowledgementAuditId: stopReceipt.acknowledgementAuditId,
+    });
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(26);
+    expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.eventType, "native.process_start_requested"))).toEqual([]);
+    expect((await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.id} <> ${run.id}`)).every(row => row.status === "failed")).toBe(true);
+    await quiet();
+  });
   it("preserves a fresh foreign controller until exact expiry and admits one process retry across two owners", async () => {
     const f = await fixture(), expiry = new Date(Date.now() + 3_000);
     const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "running",
@@ -227,6 +302,22 @@ describe("residual recovery production matrix", () => {
       eq(heartbeatRunEvents.eventType, "native.local_process_stopped")))).toHaveLength(1);
     await quiet();
   });
+  it("preserves legacy null-birth recovery with exact server identity and an absent local PID group", async () => {
+    const f = await fixture();
+    const child = spawn(process.execPath, ["-e", "console.log('legacy-group-proof')"], { detached: true, stdio: "ignore" });
+    const pid = child.pid!; await once(child, "exit");
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      runtimeMode: "native", nativeIssueId: f.issue.id, status: "failed", finishedAt: new Date(),
+      processPid: pid, processGroupId: pid, runnerProfileJson: { sessionCheckpoint: { sessionId: "legacy-exact-session" } } }).returning();
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id, phase: "committed" });
+    await appendHeartbeatRunEvent(db, { companyId: f.companyId, agentId: f.agent.id, runId: run.id,
+      eventType: "native.process_identity_recorded", payload: { targetKind: "local", hostname: os.hostname(),
+        processPid: pid, processGroupId: pid } });
+    await start();
+    expect(await getRun(run.id)).toMatchObject({ processPid: null, processGroupId: null, processStartedAt: null });
+    expect(await readNativeLocalProcessStop(db, f.companyId, run.id)).toEqual({ processPid: pid, processGroupId: pid });
+    await quiet();
+  });
   it("releases remote process identity without local birth after the exact provider stop receipt", async () => {
     const f = await fixture();
     const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
@@ -324,6 +415,30 @@ describe("residual recovery production matrix", () => {
     await start(); await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle"));
     await quiet();
   });
+  it("retains only skipped agent lock work until a lock-only transaction releases it", async () => {
+    const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
+    await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id, status: "succeeded", finishedAt: old });
+    await db.update(agents).set({ status: "running", updatedAt: old }).where(eq(agents.id, f.agent.id));
+    let release!: () => void, locked!: () => void;
+    const held = new Promise<void>(resolve => { release = resolve; });
+    const ready = new Promise<void>(resolve => { locked = resolve; });
+    const lock = db.transaction(async tx => {
+      await tx.select({ id: agents.id }).from(agents).where(eq(agents.id, f.agent.id)).for("update");
+      locked(); await held;
+    });
+    try {
+      await ready; await start();
+      expect(deadlines.get("stranded-agent-status")!()).toBe(Date.now() + 30_000);
+      expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running");
+    } finally { release(); await lock; }
+    await advance(29_999);
+    expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running");
+    await advance(1);
+    await vi.waitFor(async () => expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("idle"));
+    expect(await db.select().from(activityLog).where(and(eq(activityLog.entityId, f.agent.id),
+      eq(activityLog.action, "agent.status_reconciled")))).toHaveLength(1);
+    await quiet();
+  });
   it("moves the silence threshold on actual output and folds same-run completion only at strict snooze expiry", async () => {
     const f = await fixture(), now = new Date(), old = new Date(now.getTime() - 10 * 60_000);
     await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id));

server/src/__tests__/native-cancellation-request.integration.test.ts

@@ -2,19 +2,21 @@ import { randomUUID } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { and, eq, inArray, sql } from "drizzle-orm";
 import { agents, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { rethrowNativeCancellationLockConflict, nativeRetryCancellationCommitCondition, cancellationIntentId, claimCancellationRequest, startupCancellationFence } from "../services/native-runtime/native-cancellation-request.js";
 
 describe("atomic caller cancellation request ownership", () => {
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let db: ReturnType<typeof createDb>;
   let cancelNativeSession: typeof import("../services/native-runtime/native-session-executor.js").cancelNativeSession;
   beforeAll(async () => {
     ({ cancelNativeSession } = await import("../services/native-runtime/native-session-executor.js"));
-    database = await startEmbeddedPostgresTestDatabase("caller-stop-");
-    db = createDb(database.connectionString);
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_CANCELLATION_TEST_DATABASE_URL");
+    if (!external) database = await startEmbeddedPostgresTestDatabase("caller-stop-");
+    db = createDb(external ?? database!.connectionString, { maxConnections: 6 });
   }, 90_000);
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
   async function fixture() {
     const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID();
     await db.insert(companies).values({ id: companyId, name: "Stop claims", issuePrefix: randomUUID().slice(0, 8) });

server/src/__tests__/native-harness-backup-expiration.test.ts

@@ -24,6 +24,8 @@ import {
   companies,
   createDb,
   environmentLeases,
+  environments,
+  sandboxCreationOperations,
   heartbeatRuns,
   issues,
   nativeRunFinalizations,
@@ -768,6 +770,43 @@ describeEmbeddedPostgres("native harness backup expiration", () => {
     await quiet(); expect(await exists(own.current)).toBe(true);
   });
 
+  it.each([false, true])("sandbox attachment reconciles only its retained root using durable run authority (explicit run: %s)", async explicitRun => {
+    const own = await seedSession({ runStatus: "running" });
+    const unrelated = [];
+    for (let index = 0; index < 16; index++) unrelated.push(await seedSession({ companyId: own.companyId,
+      agentId: own.agentId, issueStatus: "todo", runStatus: index % 2 ? "running" : "succeeded" }));
+    const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "sandbox" }).returning();
+    async function attach(runId: string) {
+      const operationId = randomUUID();
+      await db.insert(sandboxCreationOperations).values({ id: operationId, companyId: own.companyId,
+        environmentId: environment!.id, originalEnvironmentId: environment!.id, runId,
+        provider: "controlled", driverKey: "controlled", config: {}, metadata: {}, nextAttemptAt: new Date() });
+      return environmentService(db).acquireLease({ creationOperationId: operationId, companyId: own.companyId,
+        environmentId: environment!.id, ...(explicitRun ? { heartbeatRunId: runId } : {}) });
+    }
+    await startBackups(); await settled();
+    const reads = vi.spyOn(fs, "readFile");
+    try {
+      const lease = await attach(own.runId); await settled();
+      expect(reads.mock.calls.map(call => String(call[0])))
+        .toEqual([path.join(own.root, "control-plane", "control-plane-state.json")]);
+      expect((await db.select().from(sandboxCreationOperations).where(eq(sandboxCreationOperations.leaseId, lease.id)))[0])
+        .toMatchObject({ state: "attached", runId: own.runId });
+      reads.mockClear();
+      const [run] = await db.insert(heartbeatRuns).values({ companyId: own.companyId, agentId: own.agentId, status: "running" }).returning();
+      await attach(run!.id); await settled();
+      expect(reads).not.toHaveBeenCalled();
+      expect(await exists(own.current)).toBe(true);
+      for (const other of unrelated) expect(await exists(other.current)).toBe(true);
+      await quiet();
+    } finally {
+      reads.mockRestore();
+      await db.delete(sandboxCreationOperations).where(eq(sandboxCreationOperations.companyId, own.companyId));
+      await db.delete(environmentLeases).where(eq(environmentLeases.environmentId, environment!.id));
+      await db.delete(environments).where(eq(environments.id, environment!.id));
+    }
+  });
+
   it("indexes protected session relations so a new run, historical task close and exact process release reactivate only its backup", async () => {
     const own = await seedSession({ closedDaysAgo: 2 });
     const other = await seedSession({ companyId: own.companyId, agentId: own.agentId, issueStatus: "todo" });

server/src/__tests__/native-settled-process-release.test.ts

@@ -66,7 +66,7 @@ suite("settled native run process identity", () => {
 
   beforeAll(async () => {
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-settled-process-");
-    db = createDb(external ?? temporary!.connectionString);
+    db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 });
     if (external) await db.execute(sql`truncate companies cascade`);
     companyId = randomUUID();
     agentId = randomUUID();
@@ -245,6 +245,17 @@ suite("settled native run process identity", () => {
     },
   );
 
+  it.each(["absent", "live", "foreign host", "known birth mismatch"])("preserves exact legacy null-birth compatibility without clearing unsafe identity: %s", async kind => {
+    const child = kind === "live" ? startDetached() : undefined;
+    if (child) children.push(child);
+    const { runId, pid } = await seed({ ...(child ? { pid: child.pid! } : {}),
+      ...(kind === "foreign host" ? { hostname: `${os.hostname()}-foreign` } : {}) });
+    await db.update(heartbeatRuns).set({ processStartedAt: kind === "known birth mismatch" ? new Date(0) : null })
+      .where(eq(heartbeatRuns.id, runId));
+    expect(await releaseSettledNativeRunProcess(db, runId)).toBe(kind === "absent" ? "released" : kind === "live" ? "pending" : "skipped");
+    expect((await readRun(runId)).processPid).toBe(kind === "absent" ? null : pid);
+    expect(await hasNativeLocalProcessStop(db, companyId, runId)).toBe(kind === "absent");
+  });
   it("keeps the identity while the process is still alive, then releases it after exit", async () => {
     const child = startDetached();
     children.push(child);

server/src/__tests__/slack-conversation-lifecycle.test.ts

@@ -451,8 +451,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
       let entered = false;
       await adapter.withIssueExecutionLock({ companyId: f.companyId, runId: f.runId, now: new Date() }, async (_locked, ports) => {
         entered = true;
-        if (operation === "cancel_wake") await ports.transaction.cancelDeferredWake({ companyId: f.companyId, wakeId, reason: "cancelled", now: new Date() });
-        if (operation === "fail_wake") await ports.transaction.failDeferredWake({ companyId: f.companyId, wakeId, now: new Date() });
+        if (operation === "cancel_wake") await ports.transaction.cancelDeferredWake({ companyId: f.companyId, wakeId, agentId: f.agentId, reason: "cancelled", now: new Date() });
+        if (operation === "fail_wake") await ports.transaction.failDeferredWake({ companyId: f.companyId, wakeId, agentId: f.agentId, now: new Date() });
         return { outcome: { kind: "released" as const }, postCommitEffects: [] };
       });
       expect(entered).toBe(true);

server/src/__tests__/task-watchdogs-classifier.test.ts

@@ -1,8 +1,6 @@
-import { describe, expect, it, vi } from "vitest";
+import { describe, expect, it } from "vitest";
 import { classifyTaskWatchdogSubtree, type TaskWatchdogClassifierIssue } from "../services/task-watchdogs.ts";
 
-import { createIncompleteTaskWatchdogScanReporter } from "../services/task-watchdog-diagnostics.ts";
-
 const companyId = "company-1";
 const sourceId = "source-1";
 const childId = "child-1";
@@ -37,39 +35,6 @@ function classify(overrides: Partial<Parameters<typeof classifyTaskWatchdogSubtr
 }
 
 describe("task watchdog subtree classifier", () => {
-  it("reports an unchanged incomplete scan once across startup and a day of periodic ticks", () => {
-    const warn = vi.fn();
-    const report = createIncompleteTaskWatchdogScanReporter(warn);
-    report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "startup");
-    for (let tick = 0; tick < 2880; tick += 1) {
-      report({ incomplete: 2, incompleteIssueIds: [childId, sourceId] }, "periodic");
-    }
-    expect(warn).toHaveBeenCalledTimes(1);
-    expect(warn).toHaveBeenCalledWith(
-      { incomplete: 2, newIncompleteIssueIds: [sourceId, childId] },
-      "startup task-watchdog reconciliation found incomplete subtree scans",
-    );
-  });
-
-  it("reports new incomplete issues and reports a resolved issue if it becomes incomplete again", () => {
-    const warn = vi.fn();
-    const report = createIncompleteTaskWatchdogScanReporter(warn);
-    report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "startup");
-    report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "periodic");
-    expect(warn).toHaveBeenLastCalledWith(
-      { incomplete: 2, newIncompleteIssueIds: [childId] },
-      "periodic task-watchdog reconciliation found incomplete subtree scans",
-    );
-    report({ incomplete: 0, incompleteIssueIds: [] }, "periodic");
-    expect(warn).toHaveBeenCalledTimes(2);
-    report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "periodic");
-    expect(warn).toHaveBeenCalledTimes(3);
-    expect(warn).toHaveBeenLastCalledWith(
-      { incomplete: 1, newIncompleteIssueIds: [sourceId] },
-      "periodic task-watchdog reconciliation found incomplete subtree scans",
-    );
-  });
-
   it("does not produce a stopped fingerprint from an incomplete scan", () => {
     const result = classify({ subtreeIncomplete: true });
     expect(result.state).toBe("incomplete");

server/src/services/environments.ts

@@ -68,7 +68,7 @@ export async function attachSandboxCreation(db: Db, id: string, scope: { company
     }
     if (!operation || operation.state !== "pending") throw new Error("Sandbox creation ownership has already transferred");
     const lease = await attach(tx as unknown as Db);
-    await notifyEnvironmentCleanup(tx, { companyId: scope.companyId });
+    await notifyEnvironmentCleanup(tx, { companyId: scope.companyId, runId: operation.runId });
     await tx.update(sandboxCreationOperations).set({ state: "attached", leaseId: lease.id, updatedAt: new Date() })
       .where(eq(sandboxCreationOperations.id, id));
     return lease;

server/src/services/heartbeat/agent-status.test.ts

@@ -6,23 +6,29 @@ import {
 } from "@paperclipai/db";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
 import { createHeartbeatAgentStatus, settledAgentWorkCondition } from "./agent-status.js";
 
-const support = await getEmbeddedPostgresTestSupport();
+const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_STATUS_TEST_DATABASE_URL");
+const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describePostgres = support.supported ? describe : describe.skip;
 if (!support.supported) console.warn(`Skipping agent status database tests: ${support.reason}`);
 
 describePostgres("heartbeat agent status", () => {
   let db: Db;
-  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
   let service: ReturnType<typeof createHeartbeatAgentStatus>;
   const live = new Set<string>();
   const old = () => new Date(Date.now() - 60 * 60_000);
 
   beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("paperclip-agent-status-");
-    db = createDb(database.connectionString);
+    if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-agent-status-");
+    db = createDb(external ?? database!.connectionString, { maxConnections: 6 });
+    if (external) {
+      db.$client.options.onnotice = () => {};
+      await db.execute(sql`truncate companies cascade`);
+    }
     service = createHeartbeatAgentStatus(db);
   }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
   beforeEach(async () => {
@@ -38,7 +44,7 @@ describePostgres("heartbeat agent status", () => {
     await db.delete(agents);
     await db.delete(companies);
   });
-  afterAll(async () => { await database?.cleanup(); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
 
   async function seed(status = "interrupted") {
     const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(), issueId = randomUUID();

server/src/services/heartbeat/agent-status.ts

@@ -134,7 +134,13 @@ export function createHeartbeatAgentStatus(db: Db) {
         const [locked] = await tx.select({ id: agents.id }).from(agents).where(and(
           eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),
         )).for("update", { skipLocked: true });
-        if (!locked) return null;
+        if (!locked) {
+          const [pending] = await tx.select({ id: agents.id }).from(agents).where(and(
+            eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),
+          ));
+          if (pending) scope.onDeferredDeadline?.(Date.now() + 30_000);
+          return null;
+        }
         const [current] = await tx.select().from(agents).where(and(
           eq(agents.id, candidate.id), eq(agents.companyId, candidate.companyId),
           eq(agents.status, "running"), eq(agents.lifecycleState, "ready"),

server/src/services/heartbeat/live-recovery-work.ts

@@ -1,4 +1,6 @@
-import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
+import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js";
+import { settledNativeProcessIdentityCondition } from "../native-local-process-stop.js";
+import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
 import { agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js";
@@ -68,16 +70,15 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType<typeof
               kind === "native" ? and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["running", "failed"]),
                 eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId),
                 sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`)
-                : kind === "question" ? and(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+                : kind === "question" ? and(or(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]),
+                    pendingFailedNativeRetryCancellationCondition()),
                   sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is not null`)
                   : kind === "active" ? and(eq(heartbeatRuns.status, "running"),
                     sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`)
                     : and(eq(heartbeatRuns.runtimeMode, "native"), inArray(heartbeatRuns.status, ["failed", "cancelled", "timed_out", "interrupted"]),
                       isNotNull(heartbeatRuns.finishedAt), or(isNotNull(heartbeatRuns.processPid),
                         isNotNull(heartbeatRuns.processGroupId), isNotNull(heartbeatRuns.processStartedAt)),
-                      or(isNotNull(heartbeatRuns.processStartedAt), exists(db.select({ id: environmentLeases.id })
-                        .from(environmentLeases).where(and(eq(environmentLeases.companyId, heartbeatRuns.companyId),
-                          eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), sql`${environmentLeases.provider} is distinct from 'local'`))))),
+                      settledNativeProcessIdentityCondition(db)),
             )).orderBy(asc(heartbeatRuns.id)).limit(100);
           rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, dueAt: kind === "native"
             ? Math.max(Date.now(), row.nextAttemptAt?.getTime() ?? 0, row.leaseExpiresAt?.getTime() ?? 0) : Date.now() }));

server/src/services/heartbeat/recovery.ts

@@ -1,3 +1,4 @@
+import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js";
 import { registerLiveRecoveryWork } from "./live-recovery-work.js";
 import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../native-runtime/native-environment-settlement-work.js";
 import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js";
@@ -2127,9 +2128,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
         and(
           scope.companyId ? eq(heartbeatRuns.companyId, scope.companyId) : undefined,
           scope.runId ? eq(heartbeatRuns.id, scope.runId) : undefined,
-          inArray(heartbeatRuns.status, [
+          or(inArray(heartbeatRuns.status, [
             ...CANCELLABLE_HEARTBEAT_RUN_STATUSES,
-          ]),
+          ]), pendingFailedNativeRetryCancellationCondition()),
           sql`${heartbeatRuns.contextSnapshot} -> ${NATIVE_QUESTION_CANCELLATION_CONTEXT_KEY} is not null`,
         ),
       );
@@ -2295,7 +2296,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
         runningProcesses.has(run.id) ||
         activeRunExecutions.has(run.id) ||
         coordinatorOwnedByCurrentController;
-      if (nativeRun && !locallyTracked && nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now) {
+      if (nativeRun && !locallyTracked && nativeControllerBootId && nativeControllerPid
+        && nativeControllerProcessStartedAt && nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now) {
         scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime());
         continue;
       }

server/src/services/heartbeat/retries.ts

@@ -1,6 +1,6 @@
-import { notifyRunDispatch } from "../run-dispatch-notifications.js";
-import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
 import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
+import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js";
 import { infrastructureRetryDelayMs } from "../system-recovery-state.js";
 import {
@@ -1000,8 +1000,8 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
           retryContextSnapshot.previousRunId = continuation.previousRunId;
         }
 
-        await notifyRunDispatch(tx, { companyId: run.companyId });
         await notifyAttentionRetentionWork(tx, { companyId: run.companyId });
+        await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
         const wakeupRequest = await tx
           .insert(agentWakeupRequests)
           .values({
@@ -1760,7 +1760,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
     };
 
     const updated = await db.transaction(async (tx) => {
-      await notifyRunDispatch(tx, { companyId: scheduled.run.companyId });
+      await notifyScheduledRetryDispatch(tx, { companyId: scheduled.run.companyId });
       const row = await tx
         .update(heartbeatRuns)
         .set({

server/src/services/heartbeat/run-control.ts

@@ -1242,18 +1242,19 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont
     // under lock, and the final CAS requires its own unchanged acknowledged
     // retry fence. Re-reading only retryable_failure here would strand replay.
     const pendingNativeRetry =
-      run.runtimeMode === "native" && run.status === "failed" && (
+      run.runtimeMode === "native" && run.status === "failed" && Boolean(run.nativeIssueId) && (
         caseHandoffRetryClaimed || Boolean(options.cancellationRequestId) || await db
-          .select({ runId: nativeRunFinalizations.runId })
+          .select({ phase: nativeRunFinalizations.phase, failureCode: nativeRunFinalizations.failureCode })
           .from(nativeRunFinalizations)
           .where(
             and(
               eq(nativeRunFinalizations.runId, run.id),
               eq(nativeRunFinalizations.companyId, run.companyId),
-              eq(nativeRunFinalizations.phase, "retryable_failure"),
+              eq(nativeRunFinalizations.issueId, run.nativeIssueId!),
             ),
           )
-          .then((rows) => rows.length > 0)
+          .then(([coordinator]) => nativeRetryCancellationEligible({ runId: run!.id, companyId: run!.companyId,
+            issueId: run!.nativeIssueId!, resultJson: run!.resultJson }, coordinator))
       );
     if (
       !pendingNativeRetry &&

server/src/services/heartbeat/run-output.test.ts

@@ -46,6 +46,7 @@ function createMemoryProvider() {
 // way the run row would carry it into a later attempt.
 function createRunRowDb(row: Partial<Run>) {
   return {
+    async transaction<T>(write: (tx: Db) => Promise<T>): Promise<T> { return write(this as unknown as Db); },
     update: () => ({
       set: (values: Partial<Run>) => ({
         where: async () => {
@@ -215,6 +216,7 @@ describe("object-backed diagnostic recording", () => {
     const row = { ...newRunRow };
     const writes: Array<Partial<Run>> = [];
     const db = {
+      async transaction<T>(write: (tx: Db) => Promise<T>): Promise<T> { return write(this as unknown as Db); },
       update: () => ({
         set: (values: Partial<Run>) => ({
           where: async () => {

server/src/services/issues.ts

@@ -100,6 +100,7 @@ import {
   heartbeatRuns,
   routineRuns,
   executionWorkspaces,
+  externalObjectMentions,
   issueApprovals,
   issueAccessGrants,
   issueAttachments,
@@ -12044,7 +12045,14 @@ export function issueService(db: Db) {
           if (owner) {
             await notifyIssueAttentionRetentionWork(tx, { companyId: owner.companyId, issueId: id });
             await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
-            await notifyExternalObjectRefresh(tx, owner.companyId);
+            const mentionedObjects = await tx.selectDistinct({ objectId: externalObjectMentions.objectId })
+              .from(externalObjectMentions)
+              .where(and(
+                eq(externalObjectMentions.companyId, owner.companyId),
+                eq(externalObjectMentions.sourceIssueId, id),
+                isNotNull(externalObjectMentions.objectId),
+              ));
+            for (const mention of mentionedObjects) await notifyExternalObjectRefresh(tx, owner.companyId, mention.objectId!);
             await notifyApprovalWork(tx, owner.companyId);
             await notifyStatusCardIssueChange(tx, owner.companyId);
           }

server/src/services/legacy-controller-lease.test.ts

@@ -25,7 +25,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
   let db: ReturnType<typeof createDb>;
   beforeAll(async () => {
     if (!external) database = await startEmbeddedPostgresTestDatabase("legacy-controller-");
-    db = createDb(external ?? database.connectionString);
+    db = createDb(external ?? database.connectionString, { maxConnections: 6 });
   }, 30000);
   afterAll(async () => { await closeLivenessDb(db, 1); await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
   async function seed() {
@@ -178,7 +178,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
 
   it("renews on one private pool per client that shutdown can end", async () => {
     const run = await seed();
-    const owner = createDb(external ?? database.connectionString);
+    const owner = createDb(external ?? database.connectionString, { maxConnections: 6 });
     const pool = livenessDb(owner);
     expect(pool).not.toBe(owner);
     expect(livenessDb(owner)).toBe(pool);

server/src/services/native-local-process-stop.ts

@@ -1,5 +1,5 @@
 import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
-import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
+import { and, asc, desc, eq, exists, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
 import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
 import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js";
 import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
@@ -146,6 +146,22 @@ export async function readNativeLocalProcessStop(db: Db, companyId: string, runI
 
 type SettledProcessRelease = "released" | "pending" | "skipped";
 
+export function settledNativeProcessIdentityCondition(db: Db) {
+  const identity = db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
+    eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), eq(heartbeatRunEvents.runId, heartbeatRuns.id),
+    eq(heartbeatRunEvents.eventType, PROCESS_IDENTITY_RECORDED), isNull(heartbeatRunEvents.sourceEventId),
+  )).orderBy(desc(heartbeatRunEvents.seq)).limit(1).as("settled_process_identity");
+  return or(isNotNull(heartbeatRuns.processStartedAt), exists(db.select({ id: environmentLeases.id })
+    .from(environmentLeases).where(and(eq(environmentLeases.companyId, heartbeatRuns.companyId),
+      eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), sql`${environmentLeases.provider} <> 'local'`))),
+    and(gt(heartbeatRuns.processPid, 1), eq(heartbeatRuns.processPid, heartbeatRuns.processGroupId),
+      exists(db.select({ payload: identity.payload }).from(identity).where(and(
+        sql`${identity.payload}->>'targetKind' = 'local'`, sql`${identity.payload}->>'hostname' = ${os.hostname()}`,
+        sql`${identity.payload}->>'processPid' = ${heartbeatRuns.processPid}::text`,
+        sql`${identity.payload}->>'processGroupId' = ${heartbeatRuns.processGroupId}::text`,
+      )))));
+}
+
 export async function releaseSettledNativeRunProcess(db: Db, runId: string, options: { onDeferredDeadline?: (at: number) => void } = {}): Promise<SettledProcessRelease> {
   const outcome = await db.transaction(async transaction => {
     const tx = transaction as unknown as Db;
@@ -174,7 +190,8 @@ export async function releaseSettledNativeRunProcess(db: Db, runId: string, opti
       )).orderBy(desc(heartbeatRunEvents.seq)).limit(1);
       if (identity?.payload?.targetKind !== "local" || identity.payload.hostname !== os.hostname() ||
           identity.payload.processPid !== run.processPid || identity.payload.processGroupId !== run.processGroupId ||
-          !run.processStartedAt || identity.payload.processStartedAt !== run.processStartedAt.toISOString()) return "skipped";
+          (run.processStartedAt ? identity.payload.processStartedAt !== run.processStartedAt.toISOString()
+            : !Number.isSafeInteger(run.processPid) || run.processPid! <= 1 || run.processGroupId !== run.processPid)) return "skipped";
       if (!(await recordNativeLocalProcessStop(tx, run))) {
         options.onDeferredDeadline?.(Date.now() + 30_000);
         return "pending";

server/src/services/native-runtime/native-cancellation-request.ts

@@ -41,6 +41,32 @@ export function nativeRetryCancellationEligible(input: {
       && !!native.acknowledgementAuditId && native.acknowledgementAuditId !== native.intentAuditId));
 }
 
+export function pendingFailedNativeRetryCancellationCondition() {
+  const native = sql`${heartbeatRuns.resultJson}->'nativeCancellation'`;
+  const continuingStop = sql`${nativeRunFinalizations.phase} = 'terminal_failure'
+    and ${nativeRunFinalizations.failureCode} = 'native_retry_cancelled'
+    and ${native}->>'schema' = 'paperclip.native-cancellation.v1'
+    and jsonb_typeof(${native}->'intentId') = 'string' and length(${native}->>'intentId') > 0
+    and ${native}->>'runId' = ${heartbeatRuns.id}::text
+    and ${native}->>'companyId' = ${heartbeatRuns.companyId}::text
+    and ${native}->>'issueId' = ${heartbeatRuns.nativeIssueId}::text
+    and ${native}->>'scope' = 'run'
+    and ${native}->>'dispatchState' in ('pending', 'acknowledged')
+    and jsonb_typeof(${native}->'dispatched') = 'boolean'
+    and jsonb_typeof(${native}->'intentAuditId') = 'string' and length(${native}->>'intentAuditId') > 0
+    and (${native}->>'dispatchState' = 'pending' or (
+      jsonb_typeof(${native}->'acknowledgementAuditId') = 'string'
+      and length(${native}->>'acknowledgementAuditId') > 0
+      and ${native}->>'acknowledgementAuditId' <> ${native}->>'intentAuditId'))`;
+  return sql`${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed'
+    and exists (select 1 from ${nativeRunFinalizations}
+      where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
+        and ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
+        and ${nativeRunFinalizations.issueId} = ${heartbeatRuns.nativeIssueId}
+        and (${nativeRunFinalizations.phase} = 'retryable_failure' or (${continuingStop}))
+        and ${nativeRunFinalizations.resultId} is null)`;
+}
+
 /** Final failed -> cancelled CAS: a later board/recovery outcome must win.
  * The row-locking subquery prevents a coordinator-only writer from changing
  * the retry disposition between this check and the status update. */

server/src/services/native-runtime/native-provider-capacity-retry.ts

@@ -1,5 +1,5 @@
-import { notifyRunDispatch } from "../run-dispatch-notifications.js";
 import { notifyAttentionRetentionWork } from "../retention-work-notifications.js";
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq } from "drizzle-orm";
 import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
@@ -51,8 +51,8 @@ export async function materializeNativeProviderCapacityRetry(input: {
     scheduledRetryAt: dueAt.toISOString(),
     executionRetryAccounting: accountingForScheduledRetry(run, NATIVE_PROVIDER_OVERLOADED_CODE, attempt),
   };
-  await notifyRunDispatch(input.tx, { companyId: run.companyId });
   await notifyAttentionRetentionWork(input.tx, { companyId: input.companyId });
+  await notifyScheduledRetryDispatch(input.tx, { companyId: run.companyId });
   const [wake] = await input.tx.insert(agentWakeupRequests).values({
     companyId: run.companyId, agentId: run.agentId, source: "automation", triggerDetail: "system",
     reason: NATIVE_PROVIDER_OVERLOADED_CODE, status: "queued", runId: retryId,

server/src/services/native-runtime/native-safe-replacement.test.ts

@@ -1,3 +1,4 @@
+import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
 import { getExecutionBlocker } from "../execution-blocker.js";
 import { createRunDispatch, deriveCommentId } from "../../modules/run-dispatch/index.js";
 import { buildExecutionContinuation } from "../execution-continuation.js";
@@ -577,7 +578,12 @@ const support = externalDatabaseUrl
     });
     it("persists exactly one linked successor under competing sweepers and restarts", async () => {
       const source = await seed(2);
+      const owner = heartbeatService(db);
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
+      const worker = owner.registerRunDispatchWork(coordinator, () => true);
+      await worker.ready;
       const now = new Date();
+      try {
       await Promise.all([
         reconcileSafeNativeReplacements(db, now),
         reconcileSafeNativeReplacements(db, now),
@@ -608,6 +614,10 @@ const support = externalDatabaseUrl
             .where(eq(nativeRunFinalizations.runId, source.runId))
         )[0]?.failureDetail?.successorRunId,
       ).toBe(children[0]!.id);
+      await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(children[0]!.scheduledRetryAt!.getTime()), { timeout: 5_000 });
+      } finally {
+        await worker.stop(); await coordinator.stop(); await owner.drainActiveRunExecutions();
+      }
     });
     it("retries status publication after a crash without dispatching provider work", async () => {
       await deliverExecutionStatuses(db);

server/src/services/native-runtime/native-safe-replacement.ts

@@ -1,3 +1,4 @@
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
 import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
 import { notifyExecutionRunWork } from "../execution-control-notifications.js";
@@ -410,6 +411,7 @@ export async function reconcileSafeNativeReplacements(
           })
           .returning();
         await notifyExecutionRunWork(tx, { companyId: run.companyId });
+        await notifyScheduledRetryDispatch(tx, { companyId: run.companyId });
         await tx.insert(heartbeatRuns).values({
           id: successorRunId,
           companyId: run.companyId,

server/src/services/recovery/service.ts

@@ -1,4 +1,4 @@
-import { notifyRunDispatch } from "../run-dispatch-notifications.js";
+import { notifyScheduledRetryDispatch } from "../run-dispatch-notifications.js";
 import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
 import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -2716,7 +2716,7 @@ export function recoveryService(
     const now = new Date();
     const retryAt = readProviderQuotaRetryAt(input.latestRun, now);
     return database.transaction(async (tx) => {
-      await notifyRunDispatch(tx, { companyId: input.issue.companyId });
+      await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
       const wakeup = await tx
         .insert(agentWakeupRequests)
         .values({
@@ -3317,7 +3317,7 @@ export function recoveryService(
           created = Boolean(enqueuedRun);
         } else {
           scheduledRun = await db.transaction(async (tx) => {
-            await notifyRunDispatch(tx, { companyId: input.issue.companyId });
+            await notifyScheduledRetryDispatch(tx, { companyId: input.issue.companyId });
             const wakeup = await tx
               .insert(agentWakeupRequests)
               .values({

server/src/services/run-dispatch-notifications.ts

@@ -7,6 +7,11 @@ export async function notifyRunDispatch(transaction: object, scope?: RunDispatch
   await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined);
   const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativeProcessSessions, hint);
-  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);
 }
+
+/** Only a retry row or deadline change needs to re-read the saved retry schedule. */
+export async function notifyScheduledRetryDispatch(transaction: object, scope?: RunDispatchScope) {
+  await notifyRunDispatch(transaction, scope);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch);
+}

server/src/services/scheduled-retry-cancellation-notifications.ts

@@ -1,9 +1,9 @@
-import { notifyRunDispatch } from "./run-dispatch-notifications.js";
+import { notifyScheduledRetryDispatch } from "./run-dispatch-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
 
 export type ScheduledRetryCancellationScope = { companyId: string; runId?: string };
 
 export async function notifyScheduledRetryCancellation(transaction: object, scope?: ScheduledRetryCancellationScope) {
-  await notifyRunDispatch(transaction, scope);
+  await notifyScheduledRetryDispatch(transaction, scope);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryCancellation, scope ? JSON.stringify(scope) : "*");
 }

server/src/services/task-watchdog-diagnostics.ts

@@ -1,16 +0,0 @@
-export function createIncompleteTaskWatchdogScanReporter(
-  warn: (details: { incomplete: number; newIncompleteIssueIds: string[] }, message: string) => void,
-) {
-  let previousIncompleteIssueIds = new Set<string>();
-  return (result: { incomplete: number; incompleteIssueIds: string[] }, phase: "startup" | "periodic") => {
-    const currentIssueIds = new Set(result.incompleteIssueIds);
-    const newIncompleteIssueIds = [...currentIssueIds].filter((id) => !previousIncompleteIssueIds.has(id));
-    previousIncompleteIssueIds = currentIssueIds;
-    if (newIncompleteIssueIds.length === 0) return;
-    warn(
-      { incomplete: result.incomplete, newIncompleteIssueIds },
-      `${phase} task-watchdog reconciliation found incomplete subtree scans`,
-    );
-  };
-}
-

tests/e2e/agent-chat-projects.spec.ts

@@ -459,12 +459,15 @@ test("plan approval hands the preserved revision to an assigned project task", a
         { timeout: 60_000 },
       )
       .toBe(1);
+    // Task completion is delivered back to the chat as another real turn.
+    await idle(request, f.chatPath, 3);
     const task = (
       await json(await request.get(`/api/companies/${f.company.id}/issues`))
     )[0];
     expect(task.parentId).toBeNull();
     expect(task.projectId).toBeTruthy();
     expect(task.assigneeAgentId).toBe(f.agent.id);
+    await expect(page.getByText(`Delegated work completed: ${task.identifier}`, { exact: true })).toBeVisible();
     const plan = await json(
       await request.get(`/api/issues/${task.id}/documents/plan`),
     );

tests/e2e/fixtures/agent-chat.mjs

@@ -36,6 +36,10 @@ if (!task.conversationAgentId) {
   });
   process.exit(0);
 }
+if (ctx.wakeReason === "chat_task_completed") {
+  await comment(`Delegated work completed: ${ctx.chatCompletionUpdates.map((update) => update.identifier).join(", ")}`);
+  process.exit(0);
+}
 const comments = await api(`/issues/${task.id}/comments?order=asc`);
 const current =
   comments.find((c) => c.id === ctx.wakeCommentId) ??