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

* commit 'c8a213f29b':
  Retain workspace preservation deadlines after transient contention
  Apply predecessor migrations before external recovery qualification
  Close native admission and stranded status producer edges

# Conflicts:
#	server/src/modules/active-run-watchdog/adapters/postgres.ts
#	server/src/services/execution-control-notifications.ts
#	server/src/services/heartbeat-run-events.ts

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

@@ -4,13 +4,13 @@ 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 | Maximum saved retry/claim lease deadline. Fresh claim owner validates provider/session state. Accepted result, terminal failure, or explicit ownership gate removes the entry. |
+| 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. 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. Exact-lease dispatch does not mutate timestamps for broad-page fairness. Successful CAS hands work to the existing pending-cleanup owner. |
+| 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 and run/resource transitions | 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. 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.
@@ -23,4 +23,23 @@ The real PostgreSQL matrix uses the actual heartbeat service and all recovery, d
 
 The legacy/controller process retry executes on real JavaScript and PostgreSQL clocks. Other deadline cases install their virtual clock before scheduler registration. Installing fake timers after a real scheduler timer has armed is not a valid deadline test.
 
-The initial matrix passed12/12 in97.89s. Startup registration, scheduler-disabled and unrelated-owner failure contracts passed111/111. Final expanded receipts and current-head CI/review results are recorded in the PR.
+The expanded production matrix passed23/23 in90.02s. Earlier final17 plus extracted recovery PostgreSQL contracts passed64/64 in102.08s. Existing native identity and watchdog authority suites passed44/44. Startup registration, scheduler-disabled and unrelated-owner failure contracts passed111/111. Current-head CI/review results are recorded in the PR.
+
+
+## Final producer closure
+
+Native claim rejection has two meanings. A current task, assignment, dependency, pause hold, or agent gate is a deliberate wait. Scoped recovery parks that wait until the existing dispatch/eligibility notification boundary commits. A caught database failure keeps durable retry backoff and reaches the worker error callback. Explicit legacy aggregate callers retain their prior deferral behavior.
+
+Agent resume does not override a task that issue recovery deliberately marked Blocked while the agent was paused. The production test requires a new native-owner revalidation pass, preserves that task gate, and proves quiet again. The separate pause-hold release test proves actual native claim settlement.
+
+A run terminal commit activates resource recovery even if cleanup stops before any lease update. This fanout is in execution-control notification boundaries. Liveness-only renewal stays run-only. The watchdog's own terminal fold also publishes its exact run/resource scope.
+
+The stranded-agent predicate depends on final accounting and wake receipts as well as runs. `accountRunCost` publishes before acknowledgement, including stale acknowledgement cleanup, with no budget policy required. The canonical wake status writer publishes the known agent before mutation. Direct receipt writers in wake-queue cancellation/authorization, task Retry retirement/chat settlement, late chat authorization failure, Slack stop, tree cancellation and assignment cancellation do the same. Run-coupled queue/retry cancellation paths use execution-control notifications. Newly inserted already-terminal receipts do not remove a prior blocker and need no such signal. Assignment cancellation selects only actual unfinished receipts with the canonical idle proof; historical cancellation write semantics remain unchanged.
+
+Each producer regression starts with a worker already parked and no unrelated budget/resource transition. The final receipt controls require durable agent status settlement, no scheduler deadline/idle hold, and twenty minutes with zero SQL. The native gate and terminal-resource controls captured the original failure before the missing topic fanout was added.
+
+## Workspace preservation contention
+
+The exact-lease owner forwards an optional deadline callback through the existing preservation helper. Only PostgreSQL lock-unavailable (`55P03`) and statement-cancelled (`57014`) outcomes retain a thirty-second retry. A rolled-back write intent may cause earlier bounded reconciliation. `NativeWorkspaceFinalizationBusyError` remains an ownership refusal, and wrong-company or manual/evidence gates do not acquire periodic retries. Legacy explicit callers keep the existing boolean contract.
+
+The regression holds the real task row `FOR UPDATE` until the existing one-second lock timeout fires, then releases only the transaction, without a mutation or notification. The registered worker retains its deadline and records workspace preservation exactly once on retry, handing the original allocation to stop-only pending cleanup with `retain_on_failure`. A second fixture exceeds the actual fifteen-second statement timeout with a one-shot PostgreSQL trigger. Both require the twenty-minute zero-SQL idle proof after the known producer fanout settles; wrong-company, active-claim and manual-gate controls remain parked.

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

@@ -2,12 +2,12 @@ import { randomUUID } from "node:crypto";
 import os from "node:os";
 import { spawn } from "node:child_process";
 import { once } from "node:events";
-import { setImmediate as immediate } from "node:timers/promises";
+import { setImmediate as immediate, setTimeout as realDelay } from "node:timers/promises";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { agents, authUsers, companies, companyMemberships, createDb, closeLivenessDb, closeCheckpointDb,
-  heartbeatRuns, heartbeatRunEvents, nativeRunFinalizations, environmentLeases, issueThreadInteractions,
-  issueRecoveryActions, issues, activityLog } from "@paperclipai/db";
+  heartbeatRuns, heartbeatRunEvents, nativeRunFinalizations, environmentLeases, issueThreadInteractions, applyPendingMigrations,
+  issueRecoveryActions, issues, activityLog, agentWakeupRequests } from "@paperclipai/db";
 import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
@@ -23,7 +23,13 @@ import { createHeartbeatRunOutput } from "../services/heartbeat/run-output.js";
 import { recoveryService } from "../services/recovery/service.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
 import { persistActivity } from "../services/activity-log.js";
+import { accountRunCost } from "../services/run-cost-accounting.js";
+import { createHeartbeatLifecycle } from "../services/heartbeat/run-lifecycle.js";
+import { createAgentLifecycle, configureAgentLifecycle } from "../services/agent-lifecycle.js";
+import { issueTreeControlService } from "../services/issue-tree-control.js";
+import { terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
 import { issueService } from "../services/issues.js";
+import { logger } from "../middleware/logger.js";
 
 vi.setConfig({ testTimeout: 90_000 });
 describe("residual recovery production matrix", () => {
@@ -39,6 +45,7 @@ describe("residual recovery production matrix", () => {
   beforeAll(async () => {
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LIVE_RECOVERY_TEST_DATABASE_URL");
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("live-recovery-");
+    else await applyPendingMigrations(external);
     db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
     await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
   });
@@ -51,7 +58,7 @@ describe("residual recovery production matrix", () => {
     coordinator.register = (queue, task) => { deadlines.set(queue, () => task.nextRunAt?.() ?? null); return register(queue, { ...task, onSettled: hints => { timeline.push(["settled", queue, Date.now(), [...hints ?? []]]); task.onSettled?.(hints); }, run: signal => {
       timeline.push(["pass", queue, Date.now()]);
       const pass = Promise.resolve(task.run(signal)); passes.add(pass);
-      void pass.then(() => passes.delete(pass), () => passes.delete(pass)); return pass;
+      void pass.then(() => { timeline.push(["finished", queue, Date.now()]); passes.delete(pass); }, () => passes.delete(pass)); return pass;
     } }); };
   });
   afterEach(async () => {
@@ -101,7 +108,7 @@ describe("residual recovery production matrix", () => {
     const quietAt = Date.now();
     if (vi.isFakeTimers()) await advance(31_000);
     else await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 35_000 });
-    console.info("quiet-drain", JSON.stringify({ quietAt, now: Date.now(), active: idleWorkSnapshot().active, next: [...deadlines].map(([q, at]) => [q, at()]), timeline }));
+    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;
     db.$client.options.debug = (_connection, query) => { queries.push(query); };
@@ -303,6 +310,8 @@ describe("residual recovery production matrix", () => {
       controllerBootId: randomUUID(), controllerLeaseExpiresAt: expiry }).returning();
     await db.insert(activityLog).values({ companyId: f.companyId, actorType: "agent", actorId: f.agent.id,
       agentId: f.agent.id, runId: target.id, entityType: "issue", entityId: source.id, action: "issue.updated", details: { status: "done" } });
+    const broad = await recoveryService(db, { enqueueWakeup: heartbeat.wakeup }).scanSilentActiveRuns({ companyId: f.companyId });
+    expect(broad.scanned).toBe(100); expect((await getRun(target.id)).status).toBe("running");
     await start(); await advance(31_000);
     await vi.waitFor(async () => expect((await getRun(target.id)).status).toBe("succeeded"));
     expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.status, "running"))).toHaveLength(suppressed.length);
@@ -364,6 +373,89 @@ describe("residual recovery production matrix", () => {
     });
     await quiet();
   });
+  async function preparedOrphan() {
+    const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "failed", runtimeMode: "native", nativeIssueId: f.issue.id, issueId: f.issue.id,
+      finishedAt: old, updatedAt: old }).returning();
+    const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId,
+      heartbeatRunId: run.id, issueId: f.issue.id, status: "active", provider: "daytona",
+      providerLeaseId: `sandbox-${run.id}`, metadata: { driver: "sandbox", remoteCwd: "/workspace" }, updatedAt: old }).returning();
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: {
+      schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64),
+      baselineSha256: "b".repeat(64), finalHostSha256: null, workspaceId: randomUUID(), leaseId: lease.id,
+      providerLeaseId: lease.providerLeaseId, remoteCwd: "/workspace", resourceDisposition: "destroy",
+    } } }).where(eq(heartbeatRuns.id, run.id));
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id,
+      runId: run.id, phase: "terminal_failure", recoveryState: "blocked" });
+    return { f, run, lease };
+  }
+  it.each(["task lock", "statement timeout"])("retries workspace preservation after real PostgreSQL %s without a producer mutation", async kind => {
+    const { f, run, lease } = await preparedOrphan();
+    const warnings = vi.spyOn(logger, "warn");
+    let retryAt = 0;
+    if (kind === "statement timeout") {
+      await db.execute(sql`create sequence live_preservation_slow_once`);
+      await db.execute(sql.raw(`create function live_preservation_slow_once() returns trigger language plpgsql as $$
+        begin if new.id = '${run.id}'::uuid and new.result_json->>'workspaceRestoreFailure' = 'restore_failed'
+          and nextval('live_preservation_slow_once') = 1 then perform pg_sleep(16); end if; return new; end $$`));
+      await db.execute(sql`create trigger live_preservation_slow_once before update on heartbeat_runs for each row execute function live_preservation_slow_once()`);
+      extraStops.push(async () => {
+        await db.execute(sql`drop trigger if exists live_preservation_slow_once on heartbeat_runs`);
+        await db.execute(sql`drop function if exists live_preservation_slow_once()`);
+        await db.execute(sql`drop sequence if exists live_preservation_slow_once`);
+      });
+    }
+    let release!: () => void, locked!: () => void;
+    const held = new Promise<void>(resolve => { release = resolve; });
+    const ready = new Promise<void>(resolve => { locked = resolve; });
+    const lock = kind === "task lock" ? db.transaction(async tx => {
+      await tx.select({ id: issues.id }).from(issues).where(eq(issues.id, f.issue.id)).for("update");
+      locked(); await held;
+    }) : Promise.resolve();
+    if (kind !== "task lock") locked();
+    try {
+      await ready; await start();
+      expect((await getRun(run.id)).resultJson).toBeNull();
+      expect((await getLease(lease.id)).status).toBe("active");
+      expect(warnings).toHaveBeenCalledWith(expect.objectContaining({ runId: run.id,
+        code: kind === "task lock" ? "55P03" : "57014" }), expect.any(String));
+      retryAt = deadlines.get("orphan-active-leases")!()!;
+      expect(retryAt).toBeGreaterThan(Date.now());
+      expect(retryAt).toBeLessThanOrEqual(Date.now() + 30_000);
+      if (kind === "task lock") expect(retryAt).toBe(Date.now() + 30_000);
+    } finally { release(); await lock; }
+    await advance(retryAt - Date.now() - 1); expect((await getRun(run.id)).resultJson).toBeNull();
+    await advance(1);
+    await vi.waitFor(async () => expect((await getRun(run.id)).resultJson).toMatchObject({
+      workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { leaseIds: [lease.id] },
+    }));
+    expect(await db.select().from(activityLog).where(and(eq(activityLog.runId, run.id),
+      eq(activityLog.action, "issue.workspace_recovery_required")))).toHaveLength(1);
+    expect(await getLease(lease.id)).toMatchObject({ status: "pending_cleanup", leasePolicy: "retain_on_failure" });
+    await quiet();
+    expect(await db.select().from(activityLog).where(and(eq(activityLog.runId, run.id),
+      eq(activityLog.action, "issue.workspace_recovery_required")))).toHaveLength(1);
+  });
+  it.each(["company", "claim", "manual"])("parks workspace preservation authority gates without timeout retries: %s", async gate => {
+    const { f, run, lease } = await preparedOrphan();
+    if (gate === "company") {
+      const other = await fixture();
+      await db.update(environmentLeases).set({ companyId: other.companyId }).where(eq(environmentLeases.id, lease.id));
+    } else if (gate === "claim") {
+      await db.update(nativeRunFinalizations).set({ leaseOwner: "another-workspace-owner",
+        leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, run.id));
+    } else {
+      await db.update(nativeRunFinalizations).set({ phase: "observed", recoveryState: "blocked" })
+        .where(eq(nativeRunFinalizations.runId, run.id));
+    }
+    await start();
+    expect((await getRun(run.id)).resultJson).toBeNull();
+    expect((await getLease(lease.id)).status).toBe("active");
+    expect(deadlines.get("orphan-active-leases")!()).toBeNull();
+    expect(await db.select().from(activityLog).where(eq(activityLog.action, "issue.workspace_recovery_required"))).toEqual([]);
+    await quiet();
+  });
   it("retains failed orphan work with bounded backoff after a real PostgreSQL write failure", async () => {
     const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
     const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, status: "active", provider: "local", updatedAt: old }).returning();
@@ -387,9 +479,10 @@ describe("residual recovery production matrix", () => {
   });
   it("parks suppressed worktree recovery without probes and reconstructs it on the committed feature change", async () => {
     const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
-    const settings = instanceSettingsService(db);
+    const runtimeEnv = { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "live-recovery-worktree" };
+    const settings = instanceSettingsService(db, { runtimeEnv });
     await settings.updateExperimental({ enableWorktreeRunExecution: false });
-    heartbeat = heartbeatService(db, { runtimeEnv: { PAPERCLIP_IN_WORKTREE: "true" } });
+    heartbeat = heartbeatService(db, { runtimeEnv });
     const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, status: "active", provider: "local", updatedAt: old }).returning();
     await start(); expect((await getLease(lease.id)).status).toBe("active"); await quiet();
     try {
@@ -399,6 +492,70 @@ describe("residual recovery production matrix", () => {
     } finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); }
     await quiet();
   });
+  it.each(["agent", "hold"] as const)("revalidates parked native session work on ordinary %s eligibility", async gate => {
+    const f = await fixture(), tree = issueTreeControlService(db);
+    const lifecycleWorker = configureAgentLifecycle(db, { requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete" });
+    extraStops.push(() => lifecycleWorker.stop());
+    await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issue.id));
+    const actor = { actorType: "user" as const, actorId: f.actorId, userId: f.actorId };
+    const hold = gate === "hold" ? await tree.createHold(f.companyId, f.issue.id, { mode: "pause", actor }) : null;
+    if (gate === "agent") { await createAgentLifecycle(db).pauseAgent(f.agent.id); await lifecycleWorker.process(f.agent.id); }
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "failed", runtimeMode: "native", nativeIssueId: f.issue.id, contextSnapshot: { issueId: f.issue.id },
+      finishedAt: new Date(), errorCode: "saved_native_failure" }).returning();
+    await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, f.issue.id));
+    await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issue.id, runId: run.id,
+      phase: "retryable_failure", nextAttemptAt: new Date(Date.now() - 1_000) });
+    await start(); await quiet();
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0].phase).toBe("retryable_failure");
+    const before = timeline.filter(event => Array.isArray(event) && event[0] === "finished" && event[1] === "native-process-sessions").length;
+    if (gate === "agent") { await createAgentLifecycle(db).resumeAgent(f.agent.id); await lifecycleWorker.process(f.agent.id); }
+    else await tree.releaseHold(f.companyId, f.issue.id, hold!.hold.id, { actor, reason: "Resume native recovery" });
+    await advance(1_000);
+    const deadline = performance.now() + 5_000;
+    while (timeline.filter(event => Array.isArray(event) && event[0] === "finished" && event[1] === "native-process-sessions").length <= before && performance.now() < deadline)
+      await realDelay(5);
+    expect(timeline.filter(event => Array.isArray(event) && event[0] === "finished" && event[1] === "native-process-sessions").length).toBeGreaterThan(before);
+    const expectedPhase = gate === "agent" ? "retryable_failure" : "terminal_failure";
+    expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)))[0].phase).toBe(expectedPhase);
+    if (gate === "agent") expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].status).toBe("blocked");
+    expect((await getRun(run.id)).errorCode).toBe("saved_native_failure");
+    expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
+    await quiet();
+  });
+  it("discovers an active lease from the ordinary terminal run commit before any lease write", async () => {
+    const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "running", runtimeMode: "legacy", startedAt: new Date(), updatedAt: new Date(),
+      controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: new Date(Date.now() + 60 * 60_000) }).returning();
+    const [lease] = await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: run.id,
+      status: "active", provider: "local", updatedAt: old }).returning();
+    await start(); expect(deadlines.get("orphan-active-leases")!()).toBeNull();
+    const ended = await terminalizeLegacyExecution({ db, run, status: "failed", patch: { finishedAt: new Date(), errorCode: "owner_process_failure" } });
+    expect(ended?.status).toBe("failed"); expect((await getLease(lease.id)).status).toBe("active");
+    await advance(1_000);
+    await vi.waitFor(async () => expect((await getLease(lease.id)).status).toBe("pending_cleanup"));
+    await quiet();
+  });
+  it.each(["accounting", "wake", "assignment_cancel"] as const)("settles parked agent status when its last %s receipt commits", async receipt => {
+    const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
+    await db.update(agents).set({ status: "running", updatedAt: old }).where(eq(agents.id, f.agent.id));
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agent.id,
+      status: "succeeded", finishedAt: old, costAccountingPending: receipt === "accounting",
+      usageJson: { accountingProviderWorkStarted: false } }).returning();
+    const [wake] = receipt !== "accounting" ? await db.insert(agentWakeupRequests).values({ companyId: f.companyId,
+      agentId: f.agent.id, runId: run.id, status: "claimed", source: "automation", reason: "fixture-last-receipt",
+      requestedByActorType: receipt === "assignment_cancel" ? "user" : "system", requestedByActorId: f.actorId, payload: { issueId: f.issue.id } }).returning() : [];
+    await start(); await quiet();
+    expect((await db.select().from(agents).where(eq(agents.id, f.agent.id)))[0].status).toBe("running");
+    if (receipt === "accounting") expect(await accountRunCost(db, run.id)).toBe(true);
+    else if (receipt === "assignment_cancel") await issueService(db).update(f.issue.id, { assigneeAgentId: null, assigneeUserId: f.actorId });
+    else await createHeartbeatLifecycle(db, {} as Parameters<typeof createHeartbeatLifecycle>[1])
+      .setWakeupStatus(wake.id, "completed", { finishedAt: new Date() });
+    await advance(1_000);
+    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 first producer debt during drain and warm standby, then resumes without duplicate cleanup", async () => {
     const f = await fixture(), old = new Date(Date.now() - 6 * 60_000);
     draining = true; await start();

server/src/modules/active-run-watchdog/adapters/postgres.ts

@@ -1,4 +1,4 @@
-import { notifyActiveRunSilence, notifyLiveRunRecovery } from "../../../services/live-run-recovery-notifications.js";
+import { notifyActiveRunSilence, notifyLiveRunRecovery, notifyLiveResourceRecovery } from "../../../services/live-run-recovery-notifications.js";
 import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
 import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
 import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
@@ -350,6 +350,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
 
     const transactionResult = await db.transaction(async (tx) => {
       await notifyLiveRunRecovery(tx, input.run);
+      await notifyLiveResourceRecovery(tx, input.run);
       await notifyInstructionCollection(tx, { companyId, runId: input.run.id });
       await notifyExternalAgentSettlement(tx, { companyId, runId: input.run.id });
       await notifyCancelledWorkspacePreservation(tx, { companyId, runId: input.run.id });

server/src/modules/run-dispatch/adapters/postgres.ts

@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../../../services/execution-control-notifications.js";
 import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js";
 import { hasRequiredWorkspaceRecovery } from "../../../services/workspace-restore-recovery-state.js";
 import { readNativeWorkspaceSyncReference } from "../../../services/native-runtime/native-workspace-sync.js";
@@ -687,6 +688,7 @@ export function createPostgresRunDispatchAdapter(
     | { applied: true; run: HeartbeatRun; postCommitEffects: PostCommitEffect[] }
     | { applied: false }
   > {
+    await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId });
     await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
     await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId });
     await notifyScheduledRetryCancellation(tx, { companyId: input.companyId, runId: input.runId });
@@ -732,6 +734,7 @@ export function createPostgresRunDispatchAdapter(
     tx: Db,
     input: CancelSuppressedRetryInput,
   ): Promise<CancelSuppressedRetryResult> {
+    await notifyExecutionRunWork(tx, { companyId: input.companyId, runId: input.runId });
     await notifyInstructionCollection(tx, { companyId: input.companyId, runId: input.runId });
     await notifyExternalAgentSettlement(tx, { companyId: input.companyId, runId: input.runId });
     await notifyCancelledWorkspacePreservation(tx, { companyId: input.companyId, runId: input.runId });
@@ -1018,6 +1021,7 @@ export function createPostgresRunDispatchAdapter(
     now: Date,
     providerDispatchNotStarted = false,
   ): Promise<CancelStaleQueuedRunOutcome> {
+      await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id });
       await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
       await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
       await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });

server/src/modules/wake-queue/adapters/postgres.test.ts

@@ -871,7 +871,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
       expect(candidate?.id).toBe(wakeId);
       const agent = await ports.transaction.findInvokableAgent({ companyId, agentId: foreignAgentId });
       expect(agent).toBeNull();
-      const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, now: new Date() });
+      const failed = await ports.transaction.failDeferredWake({ companyId, wakeId: candidate!.id, agentId: candidate!.agentId, now: new Date() });
       expect(failed).toBe(true);
       return { outcome: { kind: "released" as const }, postCommitEffects: [] };
     });
@@ -903,6 +903,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
       async (_locked, ports) => {
         const cancelledUnderWrongCompany = await ports.transaction.cancelDeferredWake({
           companyId: otherCompanyId,
+          agentId,
           wakeId,
           reason: "cross-company cancel attempt",
           now: new Date(),
@@ -911,6 +912,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
 
         const failedUnderWrongCompany = await ports.transaction.failDeferredWake({
           companyId: otherCompanyId,
+          agentId,
           wakeId,
           now: new Date(),
         });

server/src/modules/wake-queue/adapters/postgres.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js";
 import { notifyDeferredExecutionWork } from "../../../services/deferred-execution-notifications.js";
 import { notifyRunDispatch } from "../../../services/run-dispatch-notifications.js";
 import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
@@ -275,6 +276,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
             throw error;
           }
           const now = new Date();
+          await notifyStrandedAgentStatus(tx, { companyId, agentId: row.agentId });
           await notifyChatInboundWakeNotices(tx, { companyId, issueId });
           await tx
             .update(agentWakeupRequests)
@@ -320,7 +322,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
       return { liveNonSelfCommentIds, containedSelfAuthoredComment };
     },
 
-    async cancelDeferredWake({ companyId, wakeId, reason, now }) {
+    async cancelDeferredWake({ companyId, wakeId, agentId, reason, now }) {
+      await notifyStrandedAgentStatus(tx, { companyId, agentId });
       await notifyChatInboundWakeNotices(tx, { companyId });
       const rows = await tx
         .update(agentWakeupRequests)
@@ -328,6 +331,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .where(
           and(
             eq(agentWakeupRequests.id, wakeId),
+            eq(agentWakeupRequests.agentId, agentId),
             eq(agentWakeupRequests.companyId, companyId),
             eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
           ),
@@ -352,7 +356,8 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
       return row ? toDeferredWakeCandidate(row) : null;
     },
 
-    async failDeferredWake({ companyId, wakeId, now }) {
+    async failDeferredWake({ companyId, wakeId, agentId, now }) {
+      await notifyStrandedAgentStatus(tx, { companyId, agentId });
       await notifyChatInboundWakeNotices(tx, { companyId });
       const rows = await tx
         .update(agentWakeupRequests)
@@ -365,6 +370,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
         .where(
           and(
             eq(agentWakeupRequests.id, wakeId),
+            eq(agentWakeupRequests.agentId, agentId),
             eq(agentWakeupRequests.companyId, companyId),
             eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
           ),

server/src/modules/wake-queue/adapters/queued-comment-postgres.ts

@@ -1,3 +1,5 @@
+import { notifyStrandedAgentStatus } from "../../../services/live-run-recovery-notifications.js";
+import { notifyExecutionRunWork } from "../../../services/execution-control-notifications.js";
 import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
 import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
 import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
@@ -53,7 +55,7 @@ export type QueuedCommentQueuePostgresAdapterDeps = {
   syncCommentExternalObjectsSafely(commentId: string, tx: Db): Promise<void>;
 };
 
-function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction {
+function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps, wakeAgentId: string): QueuedCommentQueueTransaction {
   return {
     async updateCommentBody({ issueId, commentId, body, updatedAt }) {
       await notifyStatusCardIssueChange(tx, companyId);
@@ -103,6 +105,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
     },
 
     async cancelWake({ wakeId, reason, now }) {
+      await notifyStrandedAgentStatus(tx, { companyId, agentId: wakeAgentId });
       await notifyChatInboundWakeNotices(tx, { companyId, issueId });
       await tx
         .update(agentWakeupRequests)
@@ -111,6 +114,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
     },
 
     async cancelQueueRun({ queueRunId, reason, now }) {
+      await notifyExecutionRunWork(tx, { companyId, runId: queueRunId });
       await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId });
       await notifyInstructionCollection(tx, { companyId, runId: queueRunId });
       await notifyExternalAgentSettlement(tx, { companyId, runId: queueRunId });
@@ -311,7 +315,7 @@ export function createQueuedCommentIssueLockWriter(db: Db, deps: QueuedCommentQu
               .then((rows) => rows[0] ?? null)
           : null;
 
-        const transaction = buildTransaction(tx, companyId, input.issue.id, deps);
+        const transaction = buildTransaction(tx, companyId, input.issue.id, deps, wakeRow.agentId);
         const wake = toWakeRow(wakeRow);
         const queueRun = queueRunRow ? toRunRow(queueRunRow) : null;
         const activeRun = activeRunRow ? toRunRow(activeRunRow) : null;

server/src/modules/wake-queue/application/ports.ts

@@ -123,6 +123,7 @@ export interface WakeQueueTransaction {
   cancelDeferredWake(input: {
     companyId: string;
     wakeId: string;
+    agentId: string;
     reason: string;
     now: Date;
   }): Promise<boolean>;
@@ -135,7 +136,7 @@ export interface WakeQueueTransaction {
     now: Date;
   }): Promise<DeferredWakeCandidate | null>;
   /** Sets `status = 'failed'` guarded by the current `deferred_issue_execution` status. */
-  failDeferredWake(input: { companyId: string; wakeId: string; now: Date }): Promise<boolean>;
+  failDeferredWake(input: { companyId: string; wakeId: string; agentId: string; now: Date }): Promise<boolean>;
   getPauseHoldFacts(input: {
     companyId: string;
     issueId: string;

server/src/modules/wake-queue/application/use-cases.ts

@@ -191,6 +191,7 @@ async function runReleaseDrain(
       await ports.transaction.cancelDeferredWake({
         companyId: run.companyId,
         wakeId: candidate.id,
+        agentId: candidate.agentId,
         reason: "Delegation closing note refers to completed child work",
         now: input.now,
       });
@@ -212,6 +213,7 @@ async function runReleaseDrain(
         await ports.transaction.cancelDeferredWake({
           companyId: run.companyId,
           wakeId: candidate.id,
+          agentId: candidate.agentId,
           reason: "Deferred task messages now belong to the current assignee",
           now: input.now,
         });
@@ -260,6 +262,7 @@ async function runReleaseDrain(
       await ports.transaction.cancelDeferredWake({
         companyId: run.companyId,
         wakeId: candidate.id,
+        agentId: candidate.agentId,
         reason: commentAction.selfAuthored
           ? "Deferred wake contained only comments authored by the finishing run"
           : "Queued messages were discarded before promotion",
@@ -288,7 +291,7 @@ async function runReleaseDrain(
     });
 
     if (wakeOutcome.kind === "fail_not_invokable") {
-      await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, now: input.now });
+      await ports.transaction.failDeferredWake({ companyId: run.companyId, wakeId: workingCandidate.id, agentId: workingCandidate.agentId, now: input.now });
       continue;
     }
 
@@ -300,7 +303,7 @@ async function runReleaseDrain(
     if (wakeOutcome.kind === "cancel_pause_hold") {
       await ports.transaction.cancelDeferredWake({
         companyId: run.companyId,
-        wakeId: workingCandidate.id,
+        wakeId: workingCandidate.id, agentId: workingCandidate.agentId,
         reason: "Deferred wake suppressed by active subtree pause hold",
         now: input.now,
       });
@@ -374,7 +377,7 @@ async function promoteDeferredWake(
   ) {
     await ports.transaction.cancelDeferredWake({
       companyId: run.companyId,
-      wakeId: workingCandidate.id,
+      wakeId: workingCandidate.id, agentId: workingCandidate.agentId,
       reason: "Deferred execution wake no longer applies to a terminal task",
       now: input.now,
     });

server/src/services/chat-channels/failed-run-retry.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js";
 import { notifyDeferredExecutionWork, notifyTaskRetrySettlement } from "../deferred-execution-notifications.js";
 import { notifyNativeResponsePresentation } from "../native-response-presentation-notifications.js";
 import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
@@ -1923,6 +1924,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
           // Receipt existence proves the earlier wait, not current authority.
           // Retire only this unadmitted wake; an admitted run keeps its history.
           await db.transaction(async tx => {
+            await notifyStrandedAgentStatus(tx, { companyId: claimed.companyId, agentId: receipt.agentId });
             await notifyChatInboundWakeNotices(tx, {
               companyId: claimed.companyId, issueId: String(claimed.payload.issueId),
             });

server/src/services/chat-channels/slack/session-stop.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "../../live-run-recovery-notifications.js";
 import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js";
 import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js";
 
@@ -585,6 +586,7 @@ export function createSlackSessionStop(db: Db, options: SlackSessionStopOptions)
           ["queued", "deferred_issue_execution"].includes(wakeup.status) &&
           !issue.executionRunId
         ) {
+          await notifyStrandedAgentStatus(tx, { companyId: current.companyId, agentId: wakeup.agentId });
           await notifyChatInboundWakeNotices(tx, { companyId: current.companyId, issueId: issue.id });
           await tx
             .update(agentWakeupRequests)

server/src/services/deferred-execution-notifications.ts

@@ -24,6 +24,7 @@ export async function notifyTaskRetrySettlement(tx: object, scope: DeferredExecu
 }
 
 export async function notifyDeferredExecutionEligibility(tx: object, scope?: { companyId: string }) {
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeProcessSessions, scope ? JSON.stringify({ companyId: scope.companyId }) : "*");
   if (scope) {
     await notifyDeferredExecutionWork(tx, scope);
     await notifyNativeStatusWakeWork(tx, scope);

server/src/services/execution-control-notifications.ts

@@ -1,4 +1,4 @@
-import { notifyLiveRunRecovery } from "./live-run-recovery-notifications.js";
+import { notifyLiveRunRecovery, notifyLiveResourceRecovery } from "./live-run-recovery-notifications.js";
 import { notifyAttentionRetentionWork } from "./retention-work-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
@@ -16,6 +16,7 @@ export async function notifyExecutionRunWork(transaction: object, scope?: Execut
     await notifyStatusCardIssueChange(transaction, scope.companyId);
     await notifyDeferredExecutionEligibility(transaction, scope);
     await notifyLiveRunRecovery(transaction, scope);
+    await notifyLiveResourceRecovery(transaction, scope);
     await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId);
   }
   for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
@@ -32,11 +33,14 @@ export function writeExecutionControl<T>(
 ): Promise<T> {
   return db.transaction(async transaction => {
     await notifyRunDispatch(transaction, scope);
-    if (scope) await notifyStatusCardIssueChange(transaction, scope.companyId);
-    if (scope) await notifyAttentionRetentionWork(transaction, scope);
-    if (scope) await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId);
-    if (scope) await notifyDeferredExecutionEligibility(transaction, scope);
-    if (scope) await notifyLiveRunRecovery(transaction, scope);
+    if (scope) {
+      await notifyStatusCardIssueChange(transaction, scope.companyId);
+      await notifyAttentionRetentionWork(transaction, scope);
+      await notifyRoutineExecutionStatus(transaction, scope.issueId, scope.companyId);
+      await notifyDeferredExecutionEligibility(transaction, scope);
+      await notifyLiveRunRecovery(transaction, scope);
+      await notifyLiveResourceRecovery(transaction, scope);
+    }
     await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
     for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
     return write(transaction as unknown as Db);

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

@@ -78,12 +78,12 @@ export async function appendHeartbeatRunEvent(
   db: Db,
   input: AppendHeartbeatRunEventInput,
 ): Promise<AppendHeartbeatRunEventResult> {
-  const projectsRecovery = ["native.process_start_requested", "native.process_identity_recorded", "native.local_process_stopped"].includes(input.eventType);
   const projectsRetention = input.eventType === "lifecycle" && Boolean(input.message?.startsWith(RETRY_EXHAUSTION_MESSAGE_PREFIX));
+  const projectsRecovery = ["native.process_start_requested", "native.process_identity_recorded", "native.local_process_stopped"].includes(input.eventType);
   const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null;
   const projectsProgress = isSafeNativeChatProgressEventType(input.eventType);
   const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent);
-  if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRetention && !projectsRecovery) {
+  if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment && !projectsRecovery && !projectsRetention) {
     const allocated = db.$with("allocated_run_event", {}).as(sql`
       update ${heartbeatRuns}
       set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1,

server/src/services/heartbeat/queue.ts

@@ -1,3 +1,4 @@
+import { notifyExecutionRunWork } from "../execution-control-notifications.js";
 import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js";
 import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js";
 import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
@@ -872,6 +873,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
         await notifyInstructionCollection(tx, { companyId: current.companyId, runId: current.id });
         await notifyExternalAgentSettlement(tx, { companyId: current.companyId, runId: current.id });
         await notifyCancelledWorkspacePreservation(tx, { companyId: current.companyId, runId: current.id });
+        await notifyExecutionRunWork(tx, { companyId: current.companyId, agentId: current.agentId, runId: current.id });
         await notifyUnsafeWorkspaceExportRunChange(tx, current);
         [terminal] = await tx
           .update(heartbeatRuns)
@@ -998,6 +1000,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
           await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
           await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
           await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
+          await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id });
           await notifyUnsafeWorkspaceExportRunChange(tx, run);
           const [row] = await tx.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(),
             errorCode: "case_handoff_superseded", error: "This case handoff was cancelled or superseded.", updatedAt: new Date() })
@@ -1363,7 +1366,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
                   await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
                   await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
                   await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
-          await notifyUnsafeWorkspaceExportRunChange(tx, run);
+                  await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id });
+                  await notifyUnsafeWorkspaceExportRunChange(tx, run);
                   const [cancelled] = await tx
                     .update(heartbeatRuns)
                     .set({
@@ -1488,7 +1492,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
                 await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
                 await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
                 await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
-          await notifyUnsafeWorkspaceExportRunChange(tx, run);
+                await notifyExecutionRunWork(tx, { companyId: run.companyId, agentId: run.agentId, runId: run.id });
+                await notifyUnsafeWorkspaceExportRunChange(tx, run);
                 const [cancelled] = await tx
                   .update(heartbeatRuns)
                   .set({

server/src/services/heartbeat/recovery.ts

@@ -1496,6 +1496,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
     backoffMs: number;
     companyId?: string;
     leaseId?: string;
+    onDeferredDeadline?: (at: number) => void;
   }): Promise<{ recovered: number }> {
     const cutoff = new Date(Date.now() - opts.backoffMs);
 
@@ -1528,7 +1529,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
       if (await nativeWorkspaceExportPendingForLease(db, lease)) {
         if (lease.heartbeatRunId && !activeRunExecutions.has(lease.heartbeatRunId)
           && !adapterExecutionControls.has(lease.heartbeatRunId)
-          && await preserveOrphanedNativeWorkspace(db, lease)) recovered += 1;
+          && await preserveOrphanedNativeWorkspace(db, lease, { onDeferredDeadline: opts.onDeferredDeadline })) recovered += 1;
         if (!opts.leaseId) await deferOrphanedActiveLease(lease.id);
         continue;
       }
@@ -2081,6 +2082,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
               runtimeEnv.PAPERCLIP_INSTANCE_ID?.trim() || "paperclip-heartbeat",
             now,
             runIds: [...claimableNativeRunIds],
+            deferIneligible: !scope.runId,
+            onError: scope.onError,
             dispatch: (claim) => {
               const execution = executeRun(claim.runId, {
                 nativeLeaseOwner: claim.leaseOwner,

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

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "../live-run-recovery-notifications.js";
 import { notifyRunDispatch } from "../run-dispatch-notifications.js";
 import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js";
 import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
@@ -1419,12 +1420,13 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl
   ) {
     if (!wakeupRequestId) return;
     await db.transaction(async tx => {
-      const [current] = await tx.select({ companyId: agentWakeupRequests.companyId, status: agentWakeupRequests.status })
+      const [current] = await tx.select({ companyId: agentWakeupRequests.companyId, agentId: agentWakeupRequests.agentId, status: agentWakeupRequests.status })
         .from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeupRequestId)).for("update");
       if (!current || current.status === "cancelled") return;
       if (current.status !== (patch?.status ?? status) || patch?.payload !== undefined || patch?.runId !== undefined) {
         await notifyExternalConversationSettlement(tx, { companyId: current.companyId });
       }
+      await notifyStrandedAgentStatus(tx, { companyId: current.companyId, agentId: current.agentId });
       await tx.update(agentWakeupRequests)
         .set({ status, ...patch, updatedAt: new Date() })
 

server/src/services/issue-tree-control.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js";
 import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
 import { notifyRoutineExecutionStatus } from "./routine-execution-status-notifications.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
@@ -1226,6 +1227,11 @@ export function issueTreeControlService(db: Db) {
     if (issueIds.length === 0) return [];
     const now = new Date();
     return db.transaction(async tx => {
+      const affected = await tx.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests).where(and(
+        eq(agentWakeupRequests.companyId, companyId), inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
+        isNull(agentWakeupRequests.runId), inArray(sql<string | null>`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds),
+      )).for("update");
+      for (const agentId of new Set(affected.map(wake => wake.agentId))) await notifyStrandedAgentStatus(tx, { companyId, agentId });
       await notifyChatInboundWakeNotices(tx, { companyId });
       return tx
         .update(agentWakeupRequests)

server/src/services/issues.ts

@@ -1,4 +1,5 @@
-import { notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
+import { notifyStrandedAgentStatus, notifyActiveRunSilence } from "./live-run-recovery-notifications.js";
+import { finishedIdleWake } from "./idle-work-proof.js";
 import { notifyDeferredExecutionEligibility } from "./deferred-execution-notifications.js";
 import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
 import { notifyRunDispatch } from "./run-dispatch-notifications.js";
@@ -11529,6 +11530,15 @@ export function issueService(db: Db) {
           // Invalidate human direction at the common assignment boundary, including
           // plugin/service writes that do not go through HTTP run cancellation.
           // Keep the requester attribution for audit; cancellation revokes its use.
+          const affectedWakes = await tx.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests).where(and(
+            eq(agentWakeupRequests.companyId, receiptExisting.companyId), eq(agentWakeupRequests.requestedByActorType, "user"),
+            sql`(${sql.raw(finishedIdleWake("agent_wakeup_requests"))}) IS NOT TRUE`,
+            sql`coalesce(${agentWakeupRequests.payload}->>'issueId', ${agentWakeupRequests.payload}->>'taskId',
+              ${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'issueId',
+              ${agentWakeupRequests.payload}->'_paperclipWakeContext'->>'taskId') = ${id}`,
+          )).for("update");
+          for (const agentId of new Set<string>(affectedWakes.map((wake: { agentId: string }) => wake.agentId)))
+            await notifyStrandedAgentStatus(tx, { companyId: receiptExisting.companyId, agentId });
           await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId });
           await tx.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() })
             .where(and(eq(agentWakeupRequests.companyId, receiptExisting.companyId),

server/src/services/native-runtime/native-failed-workspace-recovery.ts

@@ -91,7 +91,7 @@ export async function preserveNativeFailedWorkspace(db: Db, selected: Pick<typeo
 }
 
 export async function preserveOrphanedNativeWorkspace(db: Db, lease: Pick<typeof environmentLeases.$inferSelect,
-  "companyId" | "heartbeatRunId">): Promise<boolean> {
+  "companyId" | "heartbeatRunId">, options: { onDeferredDeadline?: (at: number) => void } = {}): Promise<boolean> {
   if (!lease.heartbeatRunId) return false;
   const [run] = await db.select().from(heartbeatRuns).where(and(
     eq(heartbeatRuns.companyId, lease.companyId), eq(heartbeatRuns.id, lease.heartbeatRunId),
@@ -105,6 +105,7 @@ export async function preserveOrphanedNativeWorkspace(db: Db, lease: Pick<typeof
     if (error instanceof NativeWorkspaceFinalizationBusyError) return false;
     const code = retentionContentionCode(error);
     if (!code) throw error;
+    options.onDeferredDeadline?.(Date.now() + 30_000);
     logger.warn({ runId: run.id, code }, "Native workspace retention is waiting for its current owner");
     return false;
   }

server/src/services/native-runtime/native-finalization-reconciler.ts

@@ -81,6 +81,8 @@ export async function dispatchNativeSessionResumptions(input: {
   now?: Date;
   limit?: number;
   dispatch: (claim: NativeSessionResumeClaim) => void;
+  deferIneligible?: boolean;
+  onError?: (error: unknown) => void;
 }) {
   const claims = await claimNativeSessionResumptions(input);
   for (const claim of claims) input.dispatch(claim);
@@ -99,6 +101,8 @@ export async function claimNativeSessionResumptions(input: {
   runIds?: string[];
   now?: Date;
   limit?: number;
+  deferIneligible?: boolean;
+  onError?: (error: unknown) => void;
 }): Promise<NativeSessionResumeClaim[]> {
   const now = input.now ?? new Date();
   const candidates = await input.db
@@ -137,6 +141,7 @@ export async function claimNativeSessionResumptions(input: {
   for (const candidate of candidates) {
     const leaseOwner = `${input.runnerInstanceId}:resume:${randomUUID()}`;
     let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null;
+    let claimFailed = false;
     const claimed = await input.db.transaction(async (tx) => {
       await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
       const [issue] = await tx.select().from(issues).where(and(
@@ -353,6 +358,8 @@ export async function claimNativeSessionResumptions(input: {
       });
       return true;
     }).catch((error: unknown) => {
+      claimFailed = true;
+      input.onError?.(error);
       terminalRunToEmit = null;
       logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId },
         "failed to claim native session recovery candidate");
@@ -365,7 +372,7 @@ export async function claimNativeSessionResumptions(input: {
       void reportRunFailure(input.db, terminalRunToEmit);
     }
     if (claimed) claims.push({ runId: candidate.runId, leaseOwner });
-    else await input.db.transaction(async tx => {
+    else if (claimFailed || input.deferIneligible !== false) await input.db.transaction(async tx => {
       await tx.execute(sql`select set_config('statement_timeout', '10000', true), set_config('lock_timeout', '2000', true)`);
       await notifyLiveRunRecovery(tx, { companyId: candidate.companyId, runId: candidate.runId });
       await tx.update(nativeRunFinalizations).set({
@@ -381,6 +388,7 @@ export async function claimNativeSessionResumptions(input: {
         or(isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)),
       ));
     }).catch((error: unknown) => {
+      input.onError?.(error);
       logger.warn({ err: error, companyId: candidate.companyId, runId: candidate.runId },
         "failed to defer native session recovery candidate");
     });

server/src/services/run-cost-accounting.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js";
 import { usdToUnits, unitsToCents } from "@paperclipai/shared";
 import { and, asc, eq, inArray, or, sql } from "drizzle-orm";
 import { accountingRuntimeBaselines, budgetReservations, agentRuntimeState, agents, costEvents, heartbeatRuns, nativeRunFinalizations, issues, projects, decisionInvocations, promptInjectionInvocations, type Db } from "@paperclipai/db";
@@ -38,6 +39,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
     let [run] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).for("update");
     if (!run?.costAccountingPending || !terminalStatuses.includes(run.status)) return false;
     if (run.costAccountedAt) {
+      await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId });
       await tx.update(heartbeatRuns).set({ costAccountingPending: false }).where(eq(heartbeatRuns.id, run.id));
       await budgetServiceInTransaction(tx, publications).evaluateAffectedScopes(run.companyId, run.agentId,
         text(object(object(run.usageJson).ledgerScope).projectId));
@@ -69,6 +71,7 @@ export async function accountRunCost(db: Db, runId: string, hooks: BudgetService
     if (!preProviderFailure && object(run.usageJson).accountingReceiptReady === false) return false;
     // Acknowledgement is part of the same transaction. Budget checks inside
     // this transaction must not mistake this receipt for missing accounting.
+    await notifyStrandedAgentStatus(tx, { companyId: run.companyId, agentId: run.agentId });
     await tx.update(heartbeatRuns).set({ costAccountingPending: false, costAccountedAt: new Date() }).where(eq(heartbeatRuns.id, run.id));
     await tx.update(budgetReservations).set({ state: preProviderFailure ? "released" : "settled", settledAt: new Date() })
       .where(and(eq(budgetReservations.companyId, run.companyId), eq(budgetReservations.runId, run.id), eq(budgetReservations.state, "held")));

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

@@ -6,6 +6,7 @@ export type RunDispatchScope = { companyId: string };
 export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
   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);
 }

server/src/services/task-retry.ts

@@ -1,3 +1,4 @@
+import { notifyStrandedAgentStatus } from "./live-run-recovery-notifications.js";
 import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
 import { writeExecutionControl } from "./execution-control-notifications.js";
 import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
@@ -183,6 +184,12 @@ export async function retireSupersededTaskRetry(db: Db, input: {
       sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${input.issueId}`),
   )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1);
   if (latest?.id === input.sourceRunId) return false;
+  const [wake] = await db.select({ agentId: agentWakeupRequests.agentId }).from(agentWakeupRequests).where(and(
+    eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.requestId),
+    inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued", "claimed"]),
+  )).for("update");
+  if (!wake) return false;
+  await notifyStrandedAgentStatus(db, { companyId: input.companyId, agentId: wake.agentId });
   await notifyExternalConversationSettlement(db, input);
   const retired = await db.update(agentWakeupRequests).set({ status: "skipped", error: "task_retry_superseded",
     finishedAt: new Date(), updatedAt: new Date() }).where(and(
@@ -375,6 +382,7 @@ export async function reconcileTaskChatRetryReceipts(db: Db, scope: { companyId?
       updatedAt: new Date(),
     }).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution")));
     if (receipt?.runId || failed) await db.transaction(async tx => {
+      await notifyStrandedAgentStatus(tx, { companyId: wake.companyId, agentId: wake.agentId });
       await notifyExternalConversationSettlement(tx, { companyId: wake.companyId,
         issueId: typeof wake.payload?.issueId === "string" ? wake.payload.issueId : undefined });
       await write(tx);