merge: bring retry module up to current master

Keep the retry module wired through the heartbeat service after the service split.

Co-authored-by: Paperclip <noreply@paperclip.ing>

doc/DATABASE.md

@@ -681,3 +681,46 @@ and are atomically deleted before code exchange. The `code_verifier` column hold
 this non-secret binding for this namespace; Slack bot installation does not use
 PKCE. Removal and manual recovery invalidate outstanding attempts under the same
 credential-mutation lease used by configuration.
+
+
+## Transaction-aware delivery work signals
+
+Five existing delivery queues use process-local work signals to avoid empty
+polling. Register work **before** writing a queue row, on that row's transaction:
+
+```ts
+await db.transaction(async tx => {
+  await signalDatabaseWork(tx, "queue-topic");
+  await tx.insert(existingQueueTable).values(row);
+});
+```
+
+`createDb` instruments transaction callbacks and nested savepoints. Subscribers
+receive intent immediately and settlement only after the outer transaction
+finishes. Caller-owned transactions therefore need no extra post-commit wrapper.
+Dedicated clients from `withDedicatedDbConnection` share the owner's signal
+scope. No schema changes, triggers, dedicated listener connection, or periodic
+queries are installed by this API.
+
+The first registered write in a transaction fetches `pg_current_xact_id()`.
+Awaiting registration is required: a failure at that point must prevent the
+queue write. Unrelated transactions add no queries. If the transaction rejects,
+the coordinator probes `pg_xact_status(xid)` through the root pool before scanning
+the queue. An in-progress transaction remains unresolved even when the queue
+currently looks empty. Committed/aborted results permit reconciliation; NULL
+means PostgreSQL has discarded an old, no-longer-active transaction's outcome.
+The existing durable queue supplies the work in either case. Probe failures
+retain the idle hold and schedule another attempt. Work signals never replay
+queries or convert a failed database operation into success.
+
+This uses PostgreSQL's transaction-information functions (this path requires
+PostgreSQL 14+; embedded PostgreSQL uses 18). See the
+[PostgreSQL transaction information documentation](https://www.postgresql.org/docs/18/functions-info.html#FUNCTIONS-PG-SNAPSHOT).
+The process-local intent survives coordinator replacement, but not process exit.
+Startup scans recover committed queue rows. Independent DB clients, other
+processes, late transactions from an old process, and database failover need an
+explicit ownership/wake protocol; these signals are not cross-process messaging.
+Use the same root client for in-process writers. New queue insertion paths must
+register before writing, or they can remain unseen until the next startup or
+another notification. Tests should verify both the producer and its outer
+transaction boundary.

doc/DEVELOPING.md

@@ -1290,6 +1290,16 @@ reporting and forwarding callbacks preserve construction order for the other
 heartbeat modules. Existing public helpers remain re-exported by `heartbeat.ts`.
 Keep lifecycle policy changes separate from this extraction and adapter execution.
 
+Run cancellation and cleanup are in `server/src/services/heartbeat/run-control.ts`.
+It owns Stop, agent and budget cancellation, environment lease release, and
+resumption of saved comments after execution stops. `createHeartbeatRunControl`
+binds the database, lifecycle and admission callbacks, and shared executor
+barriers without starting work. The service retains the process-wide execution
+and cancellation maps so scheduler and route instances observe the same owners.
+Forwarding callbacks preserve construction order for budget, retry, recovery,
+and queue services. Existing public helpers remain re-exported by `heartbeat.ts`.
+Keep cancellation policy changes separate from this extraction and execution.
+
 ## Wake Context Delivery
 
 Built-in adapters deliver wake context through the run prompt, including structured

doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md

@@ -0,0 +1,867 @@
+# Instance sleep and scheduler inventory
+
+Date: 2026-10-08
+
+Status: Inventory with first delivery-coordinator implementation; full idle sleep remains open
+
+Code baseline: `0ac194450`
+
+Paperclip should release idle database connections and allow inactive customer
+instances to sleep without losing scheduled work, incoming events, accounting,
+or recovery. Thousands of individually hosted instances make empty polling a
+material cost even when each query is cheap.
+
+**Architectural direction: triage each loop before choosing its replacement.**
+Work caused by a committed change or incoming event should run from that event.
+Work caused by the passage of time belongs in one in-process, instance-wide
+scheduler. A later external-waker adapter publishes only the stack's next wake
+time. Routine definitions and execution decisions remain inside the stack.
+
+Immediate delivery does not have to pass through the scheduler. Its delayed
+retry does. The same service can contain both categories; classify its work
+paths rather than treating its current polling loop as one indivisible job.
+
+The database already closes idle pooled connections after 60 seconds and opens
+new connections on demand. The primary work is eliminating unnecessary queries
+and coordinating background work, connections, and hosting sleep.
+
+## First slice: transaction-aware delivery notifications and shared scheduling
+
+Updated: 2026-10-09. This implementation covers five existing durable queues:
+
+- [x] JOB-04: feedback exports register intent before the export write.
+- [x] JOB-22: completion receipts register intent inside their transaction, including caller-owned and nested transactions. No extra post-commit action is needed from callers.
+- [x] JOB-23: connection continuations register intent before enqueue.
+- [x] JOB-25: question answers register intent before enqueue.
+- [x] JOB-24: tool receipts and review resolution register intent before enqueue.
+- [x] `createDb` centrally defers delivery wakes until the outer transaction settles. Dedicated child pools share the owning client's signal scope.
+- [x] One scheduler owns pending-work and failure retries. Empty, settled queues have no recurring deadline or database scan.
+- [x] Unresolved local writes prevent idle admission. Recovery probes PostgreSQL's transaction status before scanning the existing queue and clears the hold once both are settled.
+- [x] Tests cover nested commits, rollback, late commit/abort after a lost client response, concurrent writes during a sweep, worker replacement, restart recovery, drain/standby, and shutdown.
+- [ ] Prove the complete instance writer-admission and idle/resume boundary across all other work.
+- [ ] Migrate other actual scheduled work and expose the whole instance's earliest deadline to an external waker.
+- [ ] Verify actual app and database sleep on the hosting providers.
+
+Existing delivery rows remain authoritative; no tables or migrations are added.
+A producer adds one awaited `notifyDeliveryWork(tx, queue)` before its write.
+`createDb().transaction()` and nested savepoints are instrumented centrally. The
+first opted-in write fetches the outer transaction's XID; unrelated transactions
+add no SQL. A rejected transaction with a known XID is reconciled with
+`pg_xact_status` before the queue scan. An empty scan cannot discharge an
+in-progress transaction. Committed, aborted, or aged-out XIDs can be discharged
+after scanning. A failed status query retains the hold and a retry deadline.
+
+The scheduler retains startup sweeps, retry deadlines while durable work exists,
+and retries after failures. Idle drain permits transaction and queue checks but
+does not start deliveries; warm standby suppresses both. There is no
+unconditional 60-second recovery pass.
+Feedback flushes remain serialized and cancellable, with a 30-second upload
+limit. Vote routes return after saving rather than waiting for upload. Tool
+receipts awaiting review or execution still retain their retry cadence.
+
+**Scope:** notifications and unresolved-XID tracking are process-local to a
+`createDb` owner and its dedicated child pools. They do not observe direct SQL,
+an independently created client, or another process. Such writers need an
+explicit wake/reconciliation integration before this mechanism can support
+them. Startup scans recover already committed durable rows; they do not prove
+that transactions still running in an old process have finished. Cross-process
+handoff/fencing is separate work, as is PostgreSQL failover during uncertainty.
+
+This removes these five empty polling loops; it does not establish whole-stack
+sleep. `nextWakeAt` covers only registered queues. Database connection teardown,
+other background loops, full idle reconciliation, and external wake registration
+remain separate work.
+
+Implementation: `packages/db/src/work-signals.ts`,
+`server/src/services/work-scheduler.ts`, `delivery-work-coordinator.ts`, and
+`delivery-work-notifications.ts`.
+
+## Using this checklist
+
+Items have stable IDs so they can become separate issues or PRs. All boxes start
+unchecked: an existing mechanism or a recommendation is not proof that the
+sleep journey has passed. Mark an item complete only after its completion
+criteria have evidence, and append the issue/PR and verification result to it.
+Record deferrals with a reason and the capability that must remain awake.
+
+Source links identify files in the audited revision. Function names and stated
+cadences are navigation aids; recheck them when implementing a later revision.
+Cadences describe ordinary claimed-instance operation. Warm-standby, idle-drain,
+configuration, or feature gates can suppress individual paths. This inventory
+does not establish the actual deployed fleet configuration.
+
+## Triage before implementation
+
+Use this question for every operation: **If no new request or event arrives,
+does this operation still need to happen at a particular future time?**
+
+- **Yes: clock-driven.** Register that deadline with the internal scheduler.
+  Examples are a routine at 09:00, a retry after backoff, and a session expiry.
+- **No: event-driven or on demand.** Invoke it from the owning committed mutation,
+  verified external event, or user request. An empty queue is not a scheduled
+  task. Examples are delivering a newly committed answer and refreshing a UI
+  after an issue changes.
+
+Many existing sweeps combine both. Separate their immediate dispatch from their
+retry/timeout/recovery paths. Persistent transports and active-operation
+keepalives are lifecycle requirements, not idle business schedules; their
+ownership and shutdown need separate treatment.
+
+The tables below are a first-pass mechanism triage grounded in the inventory.
+They select where to investigate first, not permission to delete a recovery
+path. Detailed checklist items remain the implementation and verification record.
+
+### Work that genuinely needs a clock
+
+| Checklist items | Reason time matters | What belongs in the scheduler |
+| --- | --- | --- |
+| JOB-08 Agent timer heartbeats and issue monitors | The user explicitly requested future checks. | The next enabled heartbeat/monitor deadline; none when unconfigured. |
+| JOB-09 Routine cron triggers | A routine must run even without new incoming activity. | Existing `nextRunAt`, retaining catch-up and timezone semantics. |
+| JOB-07 Plugin scheduled jobs | A plugin may declare real recurring work. | Declared job deadlines; plugin load/configuration events update registration. |
+| JOB-29 Backups | A backup guarantee has a time requirement. | The next backup if this instance owns it; otherwise the explicit external backup owner. |
+| JOB-18, JOB-19 Login and setup expiry | An abandoned session/resource must expire even if nobody returns. | Actual expiry and failed-cleanup retry times, only while relevant sessions exist. |
+| JOB-11, JOB-21 Decision and secret/token expiry | Decision deadlines and some deletion/retention obligations are time-based. | Required expiry/scrub deadlines. Check validity synchronously on use regardless of whether physical cleanup ran. |
+| JOB-10 Decision retention | An unchanged item can become old enough to archive. | The next actual retention deadline; immediate change handling and notification delivery are event-driven. |
+| JOB-12, JOB-13 External objects and status cards | Some configured freshness/update policies require future work. | Only time-based refresh/evaluation deadlines; reactive updates use events. |
+| JOB-27, JOB-28 Retries and watchdogs | A retry delay or absence of expected progress can itself require action. | Concrete retry, lease-expiry, monitor, or watchdog deadlines for existing work. |
+
+### Work whose normal path should stop polling
+
+| Checklist items | Preferred normal trigger | Clock-driven remainder or qualification |
+| --- | --- | --- |
+| JOB-04 Feedback exports | Committed export enqueue. | Backoff after delivery failure and durable recovery of outstanding exports. No empty five-second scan. |
+| JOB-22 Chat completion delivery | Committed task completion/update. An existing postcommit fast path is already present. | Delivery retries and recovery; the current fast path is best-effort, so prove replacement reliability before removing its sweep. |
+| JOB-23 Connection continuation delivery | Committed connection availability/change and pending continuation enqueue. | Failed-delivery retry/claim recovery. |
+| JOB-25 Question response delivery | Committed user answer. | Retry/claim expiry while delivery is outstanding. |
+| JOB-24 Tool action reviews and deliveries | Committed approval/rejection or action enqueue. | Review expiry and delivery retries. |
+| JOB-26 Cost accounting | Arrival or durable capture of a usage receipt/result. | Recovery of unresolved receipts/reservations and failed writes. Accounting debt cannot be dropped. |
+| JOB-05 MCP and Dot events | Committed source event or subscription change. | Delivery retries and required expiry cleanup. |
+| JOB-01 Chat reconciliation | Incoming message, committed publication, or endpoint configuration change. | Split its lanes: delivery retry and missed-event recovery are temporal; socket ownership is NET work. |
+| JOB-02 Email reconciliation | Verified webhook, outgoing send, or endpoint configuration change. | Retry and provider catch-up guarantees; socket reception and catch-up are NET-01/NET-02. |
+| JOB-03 Browser-use cleanup | Browser/session lifecycle change. | Register actual idle/expiry/reconciliation deadlines while sessions exist; active leases remain protected. |
+| JOB-06 Execution-control reconciliation | Execution state transitions and committed control requests. | Timeout, lease-expiry, abandoned-operation, and failed-delivery recovery. Split the lanes before changing cadence. |
+| JOB-14 GitHub events and continuity | Verified webhook and connection lifecycle change where supported. | Credential deadlines; bounded polling only for providers/events without adequate push or replay guarantees. |
+| JOB-15 Merge confirmations | Verified merge event for an outstanding candidate. | Bounded fallback only while candidates exist and webhook/replay coverage is insufficient. |
+| JOB-16 Terminal workspaces | Committed terminal task/tree transition. | Cooldown, failed-cleanup retry, and recovery of outstanding cleanup. |
+| JOB-17 Sandbox cleanup | Resource release or failed acquire/release. | Retry/backoff and provider/fleet orphan recovery. |
+| JOB-20 Connection health | Use of the connection or configuration change. | Proactive checks only where a defined freshness guarantee requires them. |
+| JOB-10, JOB-11 Decision effects and notifications | Committed decision/relevant source mutation. | Expiry/retention stays in the clock-driven category. |
+| JOB-12, JOB-13 Reactive external objects/cards | Verified provider event or relevant local mutation. | Configured time-based refresh stays in the clock-driven category. |
+| JOB-27, JOB-28 Queued runs, assignments, and dependencies | Committed enqueue, assignment, terminal result, or dependency transition. | Time-based recovery/watchdogs stay in the clock-driven category. |
+| AUX-01, AUX-03 Telemetry and plugin log flushing | First buffered entry activates a bounded flush. | Batch delay and failed-delivery retry while nonempty; no empty heartbeat. |
+| AUX-04 Import spools and runtime status expiry | Spool creation/completion, startup, or access to an expiring in-memory value. | Schedule only required file-retention/cleanup deadlines. Lazy expiration can handle noncontractual memory cleanup. |
+| UI-01, UI-02 Stable UI data | Existing live events, initial read, foreground return, and reconnect reconciliation. | Preserve bounded fallback during a genuine realtime outage or active operation. Reconcile subscription gaps because live events are not replayed. |
+| UI-03, UI-04, UI-06 Recovery, app updates, and setup | Foreground/online/user activity and bounded setup state. | Client-side retry/update deadlines where needed; these do not become tenant server jobs. |
+
+JOB-30 is removal of the aggregate polling wrapper after its constituents are
+handled, not a new scheduled job. DB-01–DB-03 concern the connection pool;
+DB-04/DB-05 concern probes. NET-01–NET-07, UI-05, and AUX-05 concern transport or
+plugin lifetime. AUX-02/AUX-06 require checking actual exporter/SDK traffic.
+SCH, WAKE, SAFE, HOST, and TEST items are infrastructure, policy, or verification
+work rather than business jobs. This keeps every inventory item accounted for
+without making the scheduler responsible for every source of activity.
+
+### Qualification before removing a poll
+
+- [ ] **TRI-01 — Identify the real trigger and every producer.** For each
+  event-driven candidate, list the committed mutations/external events that
+  create work, including alternate API paths and recovery writers. **Complete
+  when:** every supported producer has an owned dispatch path and the service
+  needs no scan to discover normal new work.
+
+- [ ] **TRI-02 — Separate first delivery from delayed recovery.** Record which
+  paths are immediate, which have a concrete due time, and which depend on an
+  external provider's push capabilities. **Complete when:** each migrated lane
+  has an explicit trigger and its remaining deadlines are enumerated.
+
+- [ ] **TRI-03 — Prove durable delivery before retiring its backstop.** A
+  best-effort callback after commit is not a durable queue consumer. Cover
+  commit-before-signal crashes, lost signals while the process stays alive,
+  handler failures, concurrent producers, and restart. **Complete when:** an
+  outstanding record cannot become invisible indefinitely. Retain a bounded,
+  explicitly justified reconciliation path until this is proven; document its
+  wake cost rather than assuming startup recovery covers every failure.
+
+- [ ] **TRI-04 — Order work by avoidable idle traffic and migration risk.**
+  Record expected idle SQL/network reduction and the smallest focused proof for
+  the next item. **Complete when:** the next PR has one clear boundary and does
+  not require the entire scheduler or all connector lifecycles to be finished.
+
+### Initial implementation shortlist
+
+1. **DB-04/DB-05: health and probe separation.** A self-contained source of
+   avoidable demand, independent of building a scheduler. Verify actual callers
+   so the new endpoint is used.
+2. **JOB-04: event-driven feedback exports.** A contained first dispatch
+   migration. Triage its enqueue, failed-export behavior, and recovery first;
+   put only actual retry/batch deadlines in scheduling infrastructure.
+3. **JOB-22/JOB-25: completion and answer delivery.** Reuse existing postcommit
+   paths where possible, after proving that durable delivery can replace the
+   sweep rather than merely supplement it.
+4. **JOB-01/JOB-02/JOB-03: frequent empty chat, email, and browser scans.** Large
+   baseline benefit, but split the component lanes and preserve configured
+   transports, expiry, and recovery. Resource absence must be invalidated when
+   a new resource is committed.
+5. **SCH with JOB-09: routine scheduling as the first true clock-driven slice.**
+   Existing `nextRunAt` makes the real scheduler requirement concrete. Add other
+   temporal paths and then the external-waker adapter.
+
+This order supersedes starting by routing the feedback polling loop through a
+generic scheduler. Event-driven improvements can ship before that scheduler is
+complete. The resulting service still needs an owned way to retry failures.
+
+## Provider behavior and cost model
+
+| Layer | Documented behavior | Consequence |
+| --- | --- | --- |
+| Neon | Normally suspends after five minutes without active queries. Idle-in-transaction connections count as active; ordinary idle connections do not necessarily block suspension. New connection requests can reset inactivity. | Stop empty SQL polling and connection churn; do not force-close active transactions. |
+| Railway | Detailed Serverless documentation specifies five minutes without outbound packets, sampled so actual sleep is roughly 5–10 minutes. Private-network traffic and responses count. Enabling Serverless requires a deployment to affect the container. | Every recurring source of packets matters. Even a DB-free endpoint can keep the app awake when repeatedly called externally. |
+| Unikraft | `on` waits for connections/HTTP requests to finish; `idle` can suspend inactive TCP connections. Active HTTP requests prevent suspension. Stateful mode resumes a snapshot. | Quiesce connections deliberately and explicitly protect background work from premature suspension. |
+
+Provider sources: [Neon compute lifecycle](https://neon.com/docs/introduction/compute-lifecycle),
+[Neon compute management](https://neon.com/docs/manage/endpoints/),
+[Railway Serverless](https://docs.railway.com/deployments/serverless),
+[Unikraft scale to zero](https://unikraft.com/docs/features/scale-to-zero).
+
+Neon usage is reported in CU-seconds. The transcript's five-minute concern is
+primarily an idle-timeout cost, not a universal billing quantum. With a
+five-minute timeout, a trivial hourly check creates roughly an 8.3% active-time
+floor before useful work; checking every five minutes can leave compute
+continuously active. This illustration assumes the default timeout and excludes
+other traffic, processing time, and storage charges. Verify actual endpoint
+settings and the applicable plan. See [Neon usage calculations](https://neon.com/docs/introduction/usage-calculations).
+
+## 1. Internal scheduler for true deadlines
+
+- [ ] **SCH-01 — Establish the instance-wide scheduler boundary.** Route
+  application-owned delayed jobs, recurring jobs, retry deadlines, and required
+  expiry/cleanup/maintenance deadlines through one scheduler shared across all
+  companies. Immediate event/queue dispatch remains in its owning service;
+  running-work tracking is shared with sleep admission. Keep domain handlers and
+  their authorization/atomic claims in their existing services. **Complete
+  when:** one scheduling interface supports true temporal obligations, company
+  scope remains explicit, and self-hosted operation requires no external service.
+
+- [ ] **SCH-02 — Model work and deadlines explicitly.** Track a stable job/lane
+  identity, company scope where applicable, due time, pending/running state,
+  retry policy, and whether running work blocks sleep. Accept running-work
+  information from event-driven handlers without forcing their dispatch through
+  the scheduler. Distinguish scheduled
+  work, immediate queued work, waiting for an external event, and passive
+  configuration/history. **Complete when:** an empty lane has no recurring SQL
+  check, and the scheduler can explain outstanding work and the earliest due
+  time without querying the database on every timer tick.
+
+- [ ] **SCH-03 — Reconstruct state and handle committed changes.** Seed the
+  scheduler from existing durable domain records on startup; use committed
+  mutations to register, reschedule, cancel, or activate work. In-memory hints
+  cannot be the sole record of an obligation. Define how all supported writers
+  invalidate hints; initially retain the existing single-process boundary.
+  **Complete when:** restart and a crash between commit and notification recover
+  work, and creating an earlier deadline updates the scheduler immediately.
+
+- [ ] **SCH-04 — Implement local scheduling and catch-up.** Arm the nearest
+  necessary local timer and run due handlers with bounded concurrency and
+  per-lane overlap protection. Preserve each domain's timezone, missed-run,
+  coalescing, priority, and retry semantics. **Complete when:** delayed callbacks,
+  clock changes, restart, and duplicate activation cannot silently lose work or
+  bypass existing execution claims.
+
+- [ ] **SCH-05 — Expose an aggregate next wake time.** Compute the earliest
+  required deadline across the whole instance, including retries, cleanup,
+  accounting, and maintenance. Represent no scheduled work as no alarm; represent
+  unknown obligations as a sleep blocker. **Complete when:** adding, moving,
+  removing, or completing the earliest item correctly changes the aggregate.
+
+- [ ] **SCH-06 — Integrate running-work ownership and drain.** Reuse
+  [task admission](../../server/src/services/task-admission.ts),
+  [idle admission](../../server/src/middleware/idle-admission.ts), and the
+  [existing sleep protocol](../idle-sleep-safety.md). Track handler completion
+  even after an HTTP response or client disconnect. **Complete when:** draining
+  prevents new scheduled admissions and waits for accepted work without losing
+  fire-and-forget work or mutations racing the final check.
+
+- [ ] **SCH-07 — Classify transport and SDK timer exceptions.** Business
+  scheduling belongs to the scheduler. Protocol pings, socket deadlines, and
+  active-operation lease renewals can remain local to their transports, but
+  must have explicit start/stop ownership and participate in sleep blocking.
+  Browser timers use a separate foreground/activity policy. **Complete when:**
+  exceptions are inventoried and owned; no hidden independent timer can renew
+  an idle connector indefinitely after its scheduler lane is stopped.
+
+- [ ] **SCH-08 — Make scheduler migration incremental and observable.** Migrate
+  each inventory item below independently. Report pending work, next deadline,
+  active holds, and unknown/unmigrated blockers through local diagnostics.
+  Avoid periodic SQL or remote reporting merely to observe inactivity.
+  **Complete when:** an operator can explain why a stack is awake and a migrated
+  lane can be rolled back without duplicate processing. Preserve the telemetry
+  review rules if any first-party event contract changes.
+
+## 2. External waker adapter later
+
+The minimal external model is `(stack, nextWakeTime)`. Operational delivery may
+also carry a generation/idempotency value and delivery status. It must not
+require copying every routine, agent configuration, or permission into Cloud.
+The internal scheduler remains authoritative about what to execute.
+
+```mermaid
+flowchart LR
+    D[Temporal obligations and retry deadlines] --> S[Instance scheduler]
+    S --> L[Local timer while awake]
+    S -->|stack, nextWakeTime, generation| W[External waker adapter]
+    W --> G[Wake-aware ingress or provider wake]
+    U[User request or webhook] --> G
+    G -->|Reconcile due deadlines on wake| S
+    G -->|Verified request or event| H
+    L --> H[Existing domain handlers]
+    S --> H
+```
+
+- [ ] **WAKE-01 — Define the adapter and local-only behavior.** Support
+  setting/replacing and cancelling the next alarm. No external adapter is
+  required to use the internal scheduler. **Complete when:** timed work remains
+  local and the instance remains awake if no durable external wake is available;
+  introducing the internal scheduler alone never authorizes sleeping through a
+  deadline.
+
+- [ ] **WAKE-02 — Make alarm publication durable and acknowledge it before sleep.**
+  Persist publication intent/generation or use an equivalent recoverable
+  handshake. Recheck that the acknowledged alarm still covers the earliest
+  deadline before sleeping. **Complete when:** failure, lost responses, and a
+  concurrently moved-earlier deadline keep the instance awake until covered;
+  local scheduling continues during external-waker outages.
+
+- [ ] **WAKE-03 — Handle duplicate, stale, delayed, and missed wake delivery.**
+  Fence stale generations, retry failed delivery, reconcile due work after
+  wake, and retain domain-level atomic claims. **Complete when:** a repeated
+  doorbell does not duplicate work, and missed deadlines follow each domain's
+  explicit catch-up policy. Do not claim exactly-once network delivery.
+
+- [ ] **WAKE-04 — Serialize ingress, sleep, deploy, and wake.** Buffer or retry
+  incoming requests while draining; authenticate external control and preserve
+  webhook verification and acknowledgement deadlines. The host must bind its
+  final stop decision to the correct process/provider generation. **Complete
+  when:** requests arriving during sleep entry are neither lost nor admitted
+  into a process being stopped. Multi-replica support requires shared fencing.
+
+- [ ] **WAKE-05 — Implement and qualify provider adapters.** Evaluate Unikraft
+  scheduled `start` operations and a shared durable delayed-delivery mechanism
+  for Railway. Verify that provider resume actually invokes due-work processing,
+  including stateful resume without process startup. **Complete when:** an
+  isolated instance wakes for its earliest deadline and no waker polls sleeping
+  tenant databases. See [Unikraft scheduled wake-ups](https://unikraft.com/docs/features/cron-jobs).
+
+- [ ] **WAKE-06 — Protect active background execution on the host.** Connect
+  active-work holds to the provider's sleep controls. Unikraft exposes a
+  reference-counted disable mechanism; account for its signal delay. **Complete
+  when:** startup, transactions, agent control, finalization, and cleanup cannot
+  be suspended halfway through accepted work. See [Unikraft application sleep controls](https://unikraft.com/docs/tutorials/scale-to-zero-triggers).
+
+- [ ] **WAKE-07 — Validate wake concentration and cost.** Capacity-test concurrent
+  scheduled wakes, apply bounded dispatch, and use jitter only where semantics
+  permit. **Complete when:** scheduler backlog/delivery delay is observable and
+  a midnight burst does not silently skip customer jobs. Avoid periodically
+  waking all tenants just to discover whether any work exists.
+
+## 3. Database and health
+
+- [ ] **DB-01 — Qualify the existing idle pool behavior.**
+  [client.ts](../../packages/db/src/client.ts) already defaults
+  `DATABASE_IDLE_TIMEOUT_SECONDS` to 60; `0` disables reaping. Keep the existing
+  Drizzle handle and lazy reconnection; evaluate 30 seconds after pollers become
+  quiet. **Complete when:** idle connections disappear and a subsequent read,
+  write, and transaction work after actual provider suspension without per-route
+  lifecycle code. Existing tests are in
+  [client.test.ts](../../packages/db/src/client.test.ts).
+
+- [ ] **DB-02 — Preserve disconnect and transaction safety.** `.end()` permanently
+  shuts down the pool and must not become an idle operation. Reconnection does
+  not authorize replaying an interrupted write. **Complete when:** existing
+  transaction isolation after disconnect and explicitly idempotent retry rules
+  still hold; ambiguous commits are not blindly replayed. See
+  [database retry behavior](../DATABASE.md#connection-loss-and-retries).
+
+- [ ] **DB-03 — Cover every client and socket.** Better Auth shares the main DB;
+  the optional plugin migration pool uses the same idle configuration. Utility,
+  backup, and dedicated advisory-lock connections are bounded work. The audited
+  postgres.js version has a 60-second TCP keepalive; keepalive is not SQL.
+  **Complete when:** all pools can reach zero when idle, reserved/active work is
+  protected, and no new persistent LISTEN or health-ping connection is introduced
+  as a substitute for polling. Sources:
+  [DB client](../../packages/db/src/client.ts),
+  [auth adapter](../../server/src/auth/better-auth.ts),
+  [backup library](../../packages/db/src/backup-lib.ts).
+
+- [ ] **DB-04 — Split process liveness from DB readiness.** Claimed-instance
+  [health](../../server/src/routes/health.ts) executes `SELECT 1`; auth and deep
+  checks may perform additional queries. Install cheap liveness before session
+  and actor resolution. **Complete when:** normal liveness produces no SQL,
+  explicit deep readiness still tests DB health, and signed claim/bootstrap
+  mutations retain their existing validation and tracking.
+
+- [ ] **DB-05 — Audit external probes and configured endpoints.** Inspect the
+  fleet gateway, uptime monitors, deployment checks, DB pool settings, and Neon
+  suspend configuration. The [ECS example](../../docker/ecs-task-definition.json)
+  probes every 30 seconds; it is not evidence of Railway fleet behavior.
+  **Complete when:** sleeping is an expected state, repeated checks do not wake
+  tenants or reconnect to Neon, and readiness errors remain distinguishable
+  from intentional sleep. Railway's built-in health checks are deployment-only:
+  [provider documentation](https://docs.railway.com/deployments/healthchecks).
+
+## 4. Server scheduling migration inventory
+
+Apply the triage above to each operation within these services. Remove polling
+from event-driven normal paths and route only real deadlines through the internal
+scheduler. Preserve durable delivery, startup recovery, and domain-level claims.
+Longer polling intervals alone do not complete an item.
+
+- [ ] **JOB-01 — Chat reconciliation.** Current: every **1s**, with multiple
+  runtime/delivery/publication lanes and SQL even without connectors. Change:
+  disarm empty lanes and activate on committed events or deadlines. **Complete
+  when:** a tenant with no chat obligations issues no chat SQL and new inbound,
+  publication, and recovery work still runs. Sources:
+  [app.ts](../../server/src/app.ts),
+  [chat-channels.ts](../../server/src/services/chat-channels.ts).
+
+- [ ] **JOB-02 — Email reconciliation.** Current: every **1s**, querying active
+  endpoints even when none exist. Change: initialize resource/pending state and
+  update on endpoint changes, webhook receipt, sends, and retries. **Complete
+  when:** empty mail state produces no polling SQL and new work activates the
+  worker without restart. Source:
+  [email-channels.ts](../../server/src/services/email-channels.ts).
+
+- [ ] **JOB-03 — Browser-use cleanup.** Current: every **3s** with due-session
+  queries even when empty. Change: register actual session reconciliation/expiry
+  deadlines. **Complete when:** no sessions means no queries, while active leases,
+  expiry, and failed cleanup remain recoverable. Sources:
+  [app.ts](../../server/src/app.ts),
+  [browser-use.ts](../../server/src/services/browser-use.ts).
+
+- [ ] **JOB-04 — Feedback exports.** Current: every **5s**, querying the queue
+  even without a configured share client. Change: committed enqueue signals and
+  durable retries. **Complete when:** empty exports are quiet and queued exports
+  survive restart and temporary delivery failure. Sources:
+  [app.ts](../../server/src/app.ts),
+  [feedback.ts](../../server/src/services/feedback.ts).
+
+- [ ] **JOB-05 — Public MCP and Dot event dispatch.** Current: **2s** per started
+  dispatcher, including SQL-backed enablement, expiry, and delivery scans.
+  Change: cache capability state and schedule delivery/retry/expiry work.
+  **Complete when:** disabled/empty dispatchers are DB-free and pending events
+  retain subscription scope, expiry, and replay safety. Source:
+  [events.ts](../../server/src/services/public-mcp/events.ts).
+
+- [ ] **JOB-06 — Execution-control reconciliation.** Current: independent
+  **15s** recovery/delivery lanes, including when the main heartbeat is disabled.
+  Change: activate lanes from durable pending work and deadlines. **Complete
+  when:** empty lanes are quiet and abandoned finalizations, replacements,
+  continuations, status/disposition delivery, and login expiry still recover.
+  Sources: [index.ts](../../server/src/index.ts),
+  [execution-control-deadline.ts](../../server/src/services/execution-control-deadline.ts).
+
+- [ ] **JOB-07 — Plugin job scheduling.** Current: **30s** SQL even without jobs.
+  Change: register nearest `nextRunAt` and signal registration/update/removal.
+  **Complete when:** empty plugins/jobs produce no scheduler SQL and due jobs
+  retain atomic ownership. The plugin job coordinator itself is event-driven,
+  not another periodic loop. Sources:
+  [plugin-job-scheduler.ts](../../server/src/services/plugin-job-scheduler.ts),
+  [plugin-job-coordinator.ts](../../server/src/services/plugin-job-coordinator.ts).
+
+- [ ] **JOB-08 — Agent timer heartbeats and issue monitors.** Current: **30s**
+  agent/monitor scans; agent-level disabled heartbeats still require discovery.
+  Change: register each relevant deadline and aggregate locally. **Complete
+  when:** no timer work means no queries, and due work preserves pause, budget,
+  authorization, and atomic wake admission. Source:
+  [heartbeat.ts](../../server/src/services/heartbeat.ts), `tickTimers`.
+
+- [ ] **JOB-09 — Routine cron triggers.** Current: **30s** due-trigger query.
+  Change: use existing `nextRunAt` in the internal scheduler. **Complete when:**
+  timezone, catch-up/coalescing, concurrency, project pause, edits, and deletion
+  behave correctly; external waking can be added without relocating routines.
+  Source: [routines.ts](../../server/src/services/routines.ts), `tickScheduledTriggers`.
+
+- [ ] **JOB-10 — Decision retention and notifications.** Current: every **30s**,
+  builds the complete attention feed for each active company before archiving
+  and delivering notifications. Change: relevant mutation signals, actual
+  retention deadlines, and batched maintenance. **Complete when:** unchanged
+  idle companies do not rebuild feeds and required archive/notification behavior
+  remains timely. Source: [index.ts](../../server/src/index.ts), `runRetentionSweep`.
+
+- [ ] **JOB-11 — Decision expiry and recovery.** Current: **30s** scans for stuck
+  effects, continuations, and TTLs. Change: schedule their actual deadlines and
+  signal committed effects. **Complete when:** expiration remains authoritative
+  at action time and interrupted effects recover without perpetual empty scans.
+  Source: [decisions.ts](../../server/src/services/decisions.ts), `sweepExpired`.
+
+- [ ] **JOB-12 — External-object refresh.** Current: **30s** scans, including a
+  DB feature check when disabled. Change: invalidate cached capability on settings
+  changes; register `nextRefreshAt`; use webhook signals where supported.
+  **Complete when:** disabled/empty state is quiet and promised refresh freshness
+  has an owned deadline. Sources:
+  [external-objects.ts](../../server/src/services/external-objects.ts),
+  [instance-settings.ts](../../server/src/services/instance-settings.ts).
+
+- [ ] **JOB-13 — Status-card evaluation.** Current: **30s**, including a DB
+  feature check when disabled. Change: register `nextEvalAt` and relevant event
+  invalidations. **Complete when:** no capability/due cards means no queries,
+  and evaluation preserves existing assignment and cancellation behavior.
+  Source: [status-cards.ts](../../server/src/services/status-cards.ts).
+
+- [ ] **JOB-14 — GitHub events and connection continuity.** Current: **30s**
+  connection scans; provider calls depend on configured/due work. Change:
+  webhook-first activation plus explicit poll fallback, credential, and retry
+  deadlines. **Complete when:** empty state is quiet and configured connections
+  retain their delivery and continuity guarantees. Source:
+  [index.ts](../../server/src/index.ts), `scheduleGitHubConnectionEventPoll` and
+  `scheduleGitHubConnectionContinuitySweep`.
+
+- [ ] **JOB-15 — Merged pull-request confirmations.** Current: **30s** pending
+  confirmation scan, followed by GitHub checks for matching candidates. Change:
+  activate on merge events with bounded polling fallback only while candidates
+  exist. **Complete when:** confirmation resolution remains authorized and
+  restart-safe, with no empty scan. Source:
+  [issue-thread-interactions.ts](../../server/src/services/issue-thread-interactions.ts).
+
+- [ ] **JOB-16 — Terminal workspace cleanup.** Current: **30s** candidate scans.
+  Change: enqueue from terminal transitions and cooldown deadlines. **Complete
+  when:** empty state is quiet, and cleanup still checks current task state,
+  artifacts, deliveries, and ownership before deleting resources. Source:
+  [execution-workspaces.ts](../../server/src/services/execution-workspaces.ts).
+
+- [ ] **JOB-17 — Sandbox cleanup and orphan spools.** Current: **30s** scans even
+  when attempts have longer staleness/backoff. Change: durable cleanup queue with
+  next retry, immediate failed-release activation, and provider/fleet backstop.
+  **Complete when:** idle scans stop without leaking paid sandboxes or ignoring
+  malformed/unpersisted spool entries. Source:
+  [heartbeat.ts](../../server/src/services/heartbeat.ts), `sweepPendingCleanupLeases`.
+
+- [ ] **JOB-18 — Adapter login and setup-token reapers.** Current: **30s**
+  durable-session/lease scans, alongside local one-shot timers. Change: register
+  actual expiry and cleanup retry times. **Complete when:** abandoned logins and
+  interrupted teardown still release sandboxes across restart, without scans
+  when no obligations exist. Source:
+  [index.ts](../../server/src/index.ts), adapter and setup-token reaper wiring.
+
+- [ ] **JOB-19 — Custom-image setup expiry.** Current: **30s** active-session
+  expiry scans. Change: register `expiresAt` and cleanup attempts. **Complete
+  when:** no sessions means no SQL and expired provider resources are released.
+  Source: [environment-custom-images.ts](../../server/src/services/environment-custom-images.ts).
+
+- [ ] **JOB-20 — Tool connection health.** Current: **30s** DB inventory;
+  provider probes are conditional. Change: check on use/configuration change,
+  with explicit deadlines for any promised proactive health checks. **Complete
+  when:** quiet connections stop creating empty scans and failures remain
+  visible within the agreed freshness contract. Source:
+  [index.ts](../../server/src/index.ts), `tools.sweepConnectionHealth`.
+
+- [ ] **JOB-21 — Secret proposals and expired action tokens.** Current: **30s**
+  expiry/cleanup queries. Change: enforce expiry synchronously on use and
+  register required scrub/cleanup deadlines. **Complete when:** idle cleanup
+  does not extend authorization or secret-retention periods. Sources:
+  [secret-proposals.ts](../../server/src/services/secret-proposals.ts),
+  [index.ts](../../server/src/index.ts).
+
+- [ ] **JOB-22 — Chat completion delivery.** Current: **30s** outbox scan plus
+  an existing postcommit fast path. Change: extend that signal and register
+  retries only while pending. **Complete when:** committed completions survive
+  restart, duplicate activation, and provider failure without empty polling.
+  Source: [index.ts](../../server/src/index.ts).
+
+- [ ] **JOB-23 — Connection continuation delivery.** Current: **30s** outbox
+  scans. Change: committed signals, retry deadlines, and startup reconciliation.
+  **Complete when:** newly available connections resume eligible work once,
+  preserve execution gates, and have no empty SQL loop. Source:
+  [index.ts](../../server/src/index.ts).
+
+- [ ] **JOB-24 — Tool action reviews and deliveries.** Current: **30s**
+  review/session/outbox scans. Change: register review expiry and delivery
+  deadlines; activate after committed decisions. **Complete when:** approval,
+  rejection, expiry, and delivery retain authorization and replay protection.
+  Source: [index.ts](../../server/src/index.ts).
+
+- [ ] **JOB-25 — Question response delivery.** Current: **30s** outbox scans.
+  Change: committed-answer activation, durable retry deadlines, and startup
+  recovery. **Complete when:** responses preserve attempt-generation fencing
+  and recover from delivery failure without polling empty queues. Source:
+  [question-response-delivery.ts](../../server/src/services/question-response-delivery.ts).
+
+- [ ] **JOB-26 — Cost accounting and decision-model recovery.** Current:
+  **30s** reconciliation even when empty. Change: pending-receipt signals,
+  spool awareness, and durable retries. **Complete when:** no empty queries,
+  but unresolved receipts/reservations remain tracked and cannot be discarded
+  to permit sleep. Source: [heartbeat.ts](../../server/src/services/heartbeat.ts),
+  `reconcileCostAccounting`.
+
+- [ ] **JOB-27 — Run recovery, scheduled retries, and queued work.** Current:
+  **30s** orphan/retry/queue/stranded-assignment scans. Change: remain responsive
+  while work exists, register retry/control deadlines, and reconcile on startup
+  or wake. **Complete when:** quiet state has no scans and interrupted runs do
+  not duplicate provider work or lose continuation. Source:
+  [index.ts](../../server/src/index.ts), periodic heartbeat recovery chain.
+
+- [ ] **JOB-28 — Dependencies, task watchdogs, silent runs, and stale locks.**
+  Current: additional **30s** scans. Change: dependency/run transition signals,
+  active-run-only monitoring, actual timeout deadlines, and startup backstops.
+  **Complete when:** recovery remains scoped and fenced, and a completed task
+  still receives any owed watchdog review. Source:
+  [heartbeat.ts](../../server/src/services/heartbeat.ts).
+
+- [ ] **JOB-29 — Automatic database backups.** Current: enabled by default,
+  typically **hourly**, with separate dump connections. Change: register with
+  the internal scheduler; later assign durable waking or explicitly transfer
+  backup responsibility to managed infrastructure. **Complete when:** backup
+  and restore guarantees are demonstrated, active backup blocks sleep, and a
+  merely configured future backup no longer blocks indefinitely once covered
+  by an external alarm. Source: [index.ts](../../server/src/index.ts).
+
+- [ ] **JOB-30 — Replace the main tick and its disabled branch.** Current:
+  default **30s** main tick; setting `HEARTBEAT_SCHEDULER_ENABLED=false` still
+  leaves accounting, external-object, cleanup, and GitHub work, plus independent
+  workers. Change: migrate all lanes before retiring aggregate polling.
+  **Complete when:** disabled scheduling semantics remain documented and no
+  legacy fallback quietly reintroduces recurring SQL. Sources:
+  [index.ts](../../server/src/index.ts), [config.ts](../../server/src/config.ts).
+
+## 5. Connectors and transport lifetime
+
+- [ ] **NET-01 — AgentMail socket ownership.** Current: outbound WebSocket and
+  **20s** DB lease renewal; its renewal bypasses the top-level background gate.
+  Change: prefer webhook mode for hosted stacks; otherwise retain an awake
+  receiver or use shared durable ingress. **Complete when:** explicit suspend
+  closes all associated timers and message reception/replay still works.
+  Source: [email-channels.ts](../../server/src/services/email-channels.ts).
+
+- [ ] **NET-02 — AgentMail catch-up.** Current: roughly **one-minute** provider
+  reconciliation even in webhook mode. Change: catch up on wake/connect and at
+  an owned recovery deadline. **Complete when:** checkpoints and deduplication
+  survive lost webhooks, with a defined maximum recovery delay. Source:
+  [email-channels.ts](../../server/src/services/email-channels.ts).
+
+- [ ] **NET-03 — Discord Gateway.** Current: persistent connection and **5s** DB
+  ownership renewal. Change: keep unsupported stacks awake initially; evaluate
+  a shared ingress worker owning the connection, lease, and durable delivery.
+  **Complete when:** idle tenant sleep does not silently stop timely reception
+  or weaken ownership fencing. Sources:
+  [chat-channels.ts](../../server/src/services/chat-channels.ts),
+  [chat-sdk-runtime.ts](../../server/src/services/chat-sdk-runtime.ts).
+
+- [ ] **NET-04 — Photon iMessage.** Current: several gRPC subscriptions, **5s**
+  ownership renewal, and **15s** catch-up. Change: explicit transport lifecycle
+  and, where required, shared durable receiving. **Complete when:** resume
+  preserves ordered checkpoints and detects replay/history gaps; stopping the
+  top-level chat loop also retires dependent work safely. Source:
+  [receiver.ts](../../server/src/services/photon/receiver.ts).
+
+- [ ] **NET-05 — Slack webhook ingress.** Current transport already uses webhook
+  mode. Change: qualify it through wake-aware ingress rather than introducing
+  a persistent receiver. **Complete when:** cold start meets provider timing
+  through durable acknowledgement/retry where necessary, with signature
+  verification, deduplication, and company scope preserved. Source:
+  [chat-sdk-runtime.ts](../../server/src/services/chat-sdk-runtime.ts).
+
+- [ ] **NET-06 — HTTP idle connections.** Current server keepalive is **185s**
+  to outlive common proxy defaults. Change: explicitly close idle sockets during
+  drain and qualify provider-specific connection policies. **Complete when:**
+  quiet sockets do not prevent intended sleep, and normal proxy reuse does not
+  gain intermittent 502/ECONNRESET failures. Source:
+  [index.ts](../../server/src/index.ts).
+
+- [ ] **NET-07 — Active execution transports and lease renewals.** Inventory
+  runner/PRP connections, HTTP/2 bridge pings, native/legacy controller leases,
+  terminal SSH keepalives, browser viewport ownership, and per-operation claims.
+  Retain their required cadence during work. **Complete when:** active operations
+  hold the instance awake and unused transports retire; no arbitrary heartbeat
+  disabling is used to manufacture sleep. Sources:
+  [native-session-executor.ts](../../server/src/services/native-runtime/native-session-executor.ts),
+  [legacy-controller-lease.ts](../../server/src/services/legacy-controller-lease.ts),
+  [HTTP/2 bridge](../../packages/adapter-utils/src/http2-bridge-server.ts),
+  [terminal WebSocket](../../server/src/realtime/environment-custom-image-terminal-ws.ts).
+
+## 6. Browser demand
+
+These client-side schedules do not run in the server scheduler. Consolidate
+their activity policy so a parked browser does not create artificial tenant
+demand. Preserve real-time behavior while a person is actively using the app.
+
+- [ ] **UI-01 — Live WebSocket and fallback polling.** Hidden tabs already close
+  the main socket; visible tabs receive **30s** pings and use **15s** refetch
+  fallback when disconnected. Change: distinguish intentional sleep from
+  outage and decide whether unattended visible tabs may become inactive.
+  **Complete when:** reconnection does not create a self-wake loop and returning
+  users reconcile missed state. Source:
+  [LiveUpdatesProvider.tsx](../../ui/src/context/LiveUpdatesProvider.tsx).
+
+- [ ] **UI-02 — Stable page queries.** Current issue views poll at **1–3s**,
+  browser/artifact views at **3–5s**, and agents/costs/reviews/settings at
+  **5–60s**. Change: extend event invalidation and active-work-only fallback;
+  reuse [polling.ts](../../ui/src/lib/polling.ts) and
+  [useSharedPolling.ts](../../ui/src/hooks/useSharedPolling.ts).
+  **Complete when:** idle views stop artificial demand without hiding updates;
+  multiple tabs do not duplicate avoidable work. Starting sources:
+  [IssueDetail.tsx](../../ui/src/pages/IssueDetail.tsx),
+  [Costs.tsx](../../ui/src/pages/Costs.tsx),
+  [useTaskBrowsers.ts](../../ui/src/hooks/useTaskBrowsers.ts).
+
+- [ ] **UI-03 — Cloud-access error recovery.** Current transient-error retries
+  run every **5s**, explicitly in background tabs; bootstrap has faster polling.
+  Change: visibility/online gating and wake-aware recovery, with bounded
+  bootstrap behavior. **Complete when:** a hidden tab that last received 502/503
+  cannot keep waking its tenant indefinitely. Source:
+  [CloudAccessGate.tsx](../../ui/src/components/CloudAccessGate.tsx).
+
+- [ ] **UI-04 — Service-worker update checks.** Current update check is
+  **hourly**, regardless of visibility. Change: check on foreground activity or
+  serve shell/update assets from shared static hosting. **Complete when:** a
+  parked hidden tab does not periodically wake its tenant to fetch `/sw.js`.
+  Source: [service-worker-updates.ts](../../ui/src/lib/service-worker-updates.ts).
+
+- [ ] **UI-05 — Plugin SSE streams.** Current EventSource lacks visibility
+  gating and holds an HTTP response open. Change: explicit visibility/idle
+  lifetime with reconciliation or replay on reconnect. **Complete when:**
+  plugin state remains correct and abandoned streams do not block sleep.
+  Sources: [bridge.ts](../../ui/src/plugins/bridge.ts),
+  [plugin routes](../../server/src/routes/plugins.ts).
+
+- [ ] **UI-06 — Setup and enrollment polling.** OAuth/chat setup often polls at
+  **1.5–3s**, email setup at **10s**, commonly while pending. Change: keep these
+  responsive during the bounded workflow; stop on terminal state, close, or
+  inactivity with a deliberate resume path. **Complete when:** abandoned setup
+  creates neither permanent demand nor lost completion evidence. Sources:
+  [ConnectionSetupFlow.tsx](../../ui/src/features/connections/ConnectionSetupFlow.tsx),
+  [chat setup](../../ui/src/pages/apps/chat/ChatEndpointSetup.tsx),
+  [email setup](../../ui/src/pages/apps/chat/EmailEndpointSetup.tsx).
+
+## 7. Lower traffic housekeeping and plugins
+
+- [ ] **AUX-01 — First-party telemetry.** The **60s** flush returns without
+  sending when its queue is empty. Change: register flush/retry work only when
+  needed and drain or durably retain pending batches. **Complete when:** empty
+  telemetry is network-quiet and delivery semantics remain explicit. Preserve
+  generated-contract, README, and privacy-review requirements for event changes.
+  Sources: [telemetry.ts](../../server/src/telemetry.ts),
+  [telemetry client](../../packages/shared/src/telemetry/client.ts).
+
+- [ ] **AUX-02 — Optional OpenTelemetry.** No configured OTLP endpoint means a
+  no-op tracer; this is separate from first-party telemetry and the local run
+  log. Change: inspect exporter batching/connection behavior only when enabled;
+  avoid a perpetual stream of idle-maintenance spans. **Complete when:** the
+  endpoint gate remains intact and configured exporters do not defeat intended
+  sleep after their pending work settles. Source:
+  [instrumentation.ts](../../server/src/instrumentation.ts).
+
+- [ ] **AUX-03 — Plugin log buffer.** The **5s** flush avoids SQL when empty.
+  Change: schedule a bounded flush after the first buffered entry and include
+  pending writes in drain. **Complete when:** no empty timer is needed and log
+  persistence has a defined shutdown/sleep boundary. Source:
+  [plugin-host-services.ts](../../server/src/services/plugin-host-services.ts).
+
+- [ ] **AUX-04 — Import spools and in-memory status expiry.** Hourly import
+  cleanup already avoids SQL when the local spool is absent/empty. Runtime
+  status expiry is memory-only. Change: use actual deadlines or opportunistic
+  cleanup where timing is not contractual. **Complete when:** neither causes an
+  unnecessary external wake, while retained files and required cleanup remain
+  tracked. Sources:
+  [company-import-transfers.ts](../../server/src/services/company-import-transfers.ts),
+  [index.ts](../../server/src/index.ts).
+
+- [ ] **AUX-05 — Plugin worker lifecycle.** Enabled plugins can execute arbitrary
+  background work; host timers cannot prove they are quiet. Change: explicit
+  quiesce/resume, scheduled-job registration, ingress ownership, and lazy provider
+  worker startup where appropriate. **Complete when:** supported plugins prove
+  quiescence; unsupported plugins remain a named blocker. Source:
+  [plugin-worker-manager.ts](../../server/src/services/plugin-worker-manager.ts).
+
+- [ ] **AUX-06 — Verify low-priority and third-party sources at runtime.**
+  Announcements are request-driven with caching; no dedicated periodic server
+  application-update check was found. Plugin log retention has no production
+  caller in the audited revision. **Complete when:** packet traces confirm these
+  classifications and identify any SDK, gRPC, HTTP agent, monitoring, remote
+  client, or plugin traffic omitted by static inspection. Source:
+  [announcement-feed.ts](../../server/src/services/announcement-feed.ts).
+
+## 8. Sleep eligibility and alternative hosting
+
+- [ ] **SAFE-01 — Separate claimed inactivity from unclaimed warm standby.**
+  Existing warm standby bypasses SQL and recurring work only for empty,
+  unclaimed instances. Change: add claimed-instance behavior through explicit
+  scheduler/quiescence contracts. **Complete when:** an inactive claimed tenant
+  can become quiet without impersonating an unclaimed tenant or disrupting
+  signed claim handling. Source:
+  [cloud-warm-standby.ts](../../server/src/services/cloud-warm-standby.ts).
+
+- [ ] **SAFE-02 — Replace conservative blockers one capability at a time.**
+  Current safety blocks every enabled plugin, configured backups, API keys,
+  active routines/connections, nonterminal issues, and many retained records.
+  Change: distinguish current obligations from passive capability/history only
+  when a validated wake/recovery contract exists. **Complete when:** waiting for
+  a human or retaining completed history can permit sleep, while unknown state,
+  cleanup/accounting debt, and unhandled transports remain blockers. Sources:
+  [idle-sleep-safety.ts](../../server/src/services/idle-sleep-safety.ts),
+  [idle-local-work.ts](../../server/src/services/idle-local-work.ts).
+
+- [ ] **SAFE-03 — Validate cold start and stateful resume.** A snapshot preserves
+  memory, not the validity of external sockets, credentials, or leases. Change:
+  recheck deadlines and ownership on resume, recover stale connections, and
+  avoid replaying uncertain operations. **Complete when:** long sleeps,
+  expired sessions, and remote controller changes produce correct recovery
+  rather than duplicate execution or prolonged ingress rejection.
+
+- [ ] **HOST-01 — Evaluate colocated PostgreSQL separately.** Public authenticated
+  deployments currently reject an unset `DATABASE_URL` rather than falling back
+  to embedded PostgreSQL. Prototype explicit loopback PostgreSQL or deliberately
+  revise that deployment contract. **Complete when:** persistent storage,
+  backups/restores, upgrades, memory cost, and sleep/resume are qualified, with
+  agent execution isolated from the controller/database. This option does not
+  remove scheduler or wake requirements. Source:
+  [index.ts](../../server/src/index.ts), `assertCloudDatabaseContract`.
+
+## 9. Delivery order and acceptance
+
+1. Complete mechanism triage for the selected first slice: normal trigger,
+   producer coverage, durable dispatch, and remaining temporal obligations.
+2. Ship DB-free health/probe handling and contained event-driven migrations.
+   Measure empty-worker SQL reduction while preserving delivery backstops until
+   their replacements are proven. These changes do not depend on a full scheduler.
+3. Build the internal scheduler around true deadlines, starting with routines
+   and adding required retry/expiry/maintenance paths. Then add the external-waker
+   adapter and acknowledged alarm protocol before sleeping scheduled workloads.
+4. Qualify connector transports, plugin lifecycle, browser demand, and provider
+   resume behavior. Expand eligibility only with evidence.
+
+- [ ] **TEST-01 — Measure an empty claimed tenant.** After startup settles, prove
+  zero app-originated SQL and no unintended outgoing packets, then observe actual
+  Neon and hosting sleep. Use provider/control-plane metrics so measurement
+  itself does not wake the tenant DB. Compare against the original baseline.
+
+- [ ] **TEST-02 — Prove first-use recovery.** After real sleep, perform an
+  authenticated read, write, and transaction. Verify persistent results, no
+  duplicate effects, and measured end-to-end latency including authentication
+  and database startup.
+
+- [ ] **TEST-03 — Prove scheduled work.** Exercise routines, a moved-earlier
+  deadline, cancellation, retry, expiry, and maintenance through restart/sleep.
+  Deliver duplicate/stale wakes and simulate external-waker failure. Verify
+  catch-up, claims, and awake fallback.
+
+- [ ] **TEST-04 — Prove event ingress.** Deliver a verified webhook while asleep
+  and while entering sleep. Exercise a connector with its supported receiver
+  strategy. Verify acknowledgements, replay, company scope, and persistence.
+
+- [ ] **TEST-05 — Prove active-work protection.** Attempt sleep during a run,
+  transaction, backup, finalization, cleanup attempt, and pending accounting
+  receipt. Verify the intended blocker/hold and safe release after settlement.
+
+- [ ] **TEST-06 — Prove browser inactivity behavior.** Test hidden tabs,
+  unattended visible tabs, reconnect/error states, plugin streams, and multiple
+  tabs. Verify returning users see reconciled state without an accidental wake
+  storm.
+
+- [ ] **TEST-07 — Prove crash, ownership, and rollout behavior.** Test a crash
+  around commit/alarm publication, stale drain owner, provider stop racing a
+  deployment, and long stateful suspension. Roll back migrated lanes without
+  duplicate workers. Measure fleet wake concurrency and actual compute savings
+  before broad enablement.
+
+Each implementation PR should run the smallest relevant checks first and the
+repository-required verification for its handoff scope. This documentation-only
+checklist does not claim any provider sleep/wake test has passed.

packages/db/src/client.ts

@@ -3,6 +3,7 @@ import { drizzle as drizzlePg } from "drizzle-orm/postgres-js";
 import { readFile, readdir } from "node:fs/promises";
 import { fileURLToPath } from "node:url";
 import postgres from "postgres";
+import { installDatabaseWorkSignals } from "./work-signals.js";
 import * as schema from "./schema/index.js";
 
 const MIGRATIONS_FOLDER = fileURLToPath(new URL("./migrations", import.meta.url));
@@ -268,6 +269,10 @@ export async function withDedicatedDbConnection<T>(db: Db, action: (dedicated: D
 
 export function createDb(url: string, options?: DatabaseClientOptions) {
   const resolved = resolveDatabaseClientOptions(options ?? databaseClientOptionsFromEnv());
+  return createDbClient(url, resolved);
+}
+
+function createDbClient(url: string, resolved: DatabaseClientOptions, owner?: object) {
   const sql = postgres(url, postgresJsOptions(resolved));
   const key = hostPortKeyOrNull(url);
   if (key) registerClient(key, sql);
@@ -275,10 +280,10 @@ export function createDb(url: string, options?: DatabaseClientOptions) {
   // postgres.js calls that error "write CONNECTION_CLOSED" too, so the
   // message cannot establish that replay is safe. Leave retries to callers
   // that know the complete operation is idempotent.
-  const db = drizzlePg(sql, { schema });
-  dedicatedDbFactories.set(db, () => createDb(url, {
+  const db = installDatabaseWorkSignals(drizzlePg(sql, { schema }), owner);
+  dedicatedDbFactories.set(db, () => createDbClient(url, {
     ...resolved, maxConnections: 1, applicationName: "paperclip-workspace-finalization-lock",
-  }));
+  }, db));
   return db;
 }
 

packages/db/src/index.ts

@@ -42,3 +42,5 @@ export { loadWithoutEmbeddedPostgresExitHooks } from "./embedded-postgres-lifecy
 export { issueRelations } from "./schema/issue_relations.js";
 export { issueReferenceMentions } from "./schema/issue_reference_mentions.js";
 export * from "./schema/index.js";
+
+export { signalDatabaseWork, subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "./work-signals.js";

packages/db/src/work-signals.test.ts

@@ -0,0 +1,165 @@
+import { sql } from "drizzle-orm";
+import { drizzle } from "drizzle-orm/postgres-js";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { createDb, withDedicatedDbConnection } from "./client.js";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js";
+import { databaseWorkPending, installDatabaseWorkSignals, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("database work signals", () => {
+  let db: ReturnType<typeof createDb>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  beforeAll(async () => {
+    temporary = await startEmbeddedPostgresTestDatabase("work-signals-");
+    db = createDb(temporary.connectionString);
+    await db.execute(sql`create table work_signal_test (id text primary key)`);
+  }, 90_000);
+  afterAll(async () => { await temporary?.cleanup(); });
+
+  it("defers nested savepoint notifications until the outer commit and preserves return values", async () => {
+    const events: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(db, "nested", event => events.push(event));
+    try {
+      const result = await db.transaction(async tx => {
+        await tx.transaction(async nested => {
+          await signalDatabaseWork(nested, "nested");
+          await nested.execute(sql`insert into work_signal_test values ('nested')`);
+        });
+        expect(events).toEqual(["intent"]);
+        expect(databaseWorkPending(db, "nested")).toBe(true);
+        expect(await db.execute(sql`select * from work_signal_test where id = 'nested'`)).toHaveLength(0);
+        return 123;
+      });
+      expect(result).toBe(123);
+      expect(events).toEqual(["intent", "settled"]);
+      expect(databaseWorkPending(db, "nested")).toBe(false);
+      expect(await db.execute(sql`select * from work_signal_test where id = 'nested'`)).toHaveLength(1);
+    } finally { unsubscribe(); }
+  });
+
+  it("reconciles a rollback without leaving a permanent pending flag", async () => {
+    await expect(db.transaction(async tx => {
+      await signalDatabaseWork(tx, "rollback");
+      await tx.execute(sql`insert into work_signal_test values ('rollback')`);
+      throw new Error("abort");
+    })).rejects.toThrow("abort");
+    await reconcileDatabaseWork(db, "rollback");
+    expect(databaseWorkPending(db, "rollback")).toBe(false);
+    expect(await db.execute(sql`select * from work_signal_test where id = 'rollback'`)).toHaveLength(0);
+  });
+
+  it("keeps outer ownership after a caught savepoint rollback", async () => {
+    const events: string[] = [];
+    const unsubscribe = subscribeDatabaseWork(db, "savepoint", event => events.push(event));
+    try {
+      await db.transaction(async tx => {
+        await expect(tx.transaction(async nested => {
+          await signalDatabaseWork(nested, "savepoint");
+          await nested.execute(sql`insert into work_signal_test values ('savepoint')`);
+          throw new Error("rollback savepoint");
+        })).rejects.toThrow("rollback savepoint");
+        expect(databaseWorkPending(db, "savepoint")).toBe(true);
+        expect(events).toEqual(["intent"]);
+        await signalDatabaseWork(tx, "savepoint");
+        await tx.execute(sql`insert into work_signal_test values ('savepoint-retry')`);
+      });
+      expect(events).toEqual(["intent", "intent", "settled"]);
+      expect(databaseWorkPending(db, "savepoint")).toBe(false);
+    } finally { unsubscribe(); }
+  });
+
+  it.each(["commit", "abort"])("does not clear a late %s until PostgreSQL reports a settled transaction", async outcome => {
+    // Inject a lost response at the client boundary while a REAL transaction
+    // remains open on PostgreSQL. The status probe and row visibility are real.
+    const raw = drizzle(db.$client);
+    const transaction = raw.transaction.bind(raw);
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    let wrote!: () => void;
+    const written = new Promise<void>(resolve => { wrote = resolve; });
+    let serverDone!: Promise<unknown>;
+    raw.transaction = ((callback: any) => {
+      serverDone = transaction(async tx => {
+        const result = await callback(tx);
+        wrote();
+        await gate;
+        if (outcome === "abort") throw new Error("server abort");
+        return result;
+      });
+      // Observe rejection immediately; tests await the same result below.
+      void serverDone.catch(() => {});
+      return written.then(() => { throw new Error("lost transaction response"); });
+    }) as typeof raw.transaction;
+    const owner = installDatabaseWorkSignals(raw);
+    const topic = `late-${outcome}`;
+    try {
+      await expect(owner.transaction(async tx => {
+        await signalDatabaseWork(tx, topic);
+        await tx.execute(sql`insert into work_signal_test values (${topic})`);
+      })).rejects.toThrow("lost transaction response");
+      await reconcileDatabaseWork(owner, topic);
+      expect(databaseWorkPending(owner, topic)).toBe(true);
+      expect(await db.execute(sql`select * from work_signal_test where id = ${topic}`)).toHaveLength(0);
+      release(); await serverDone.catch(() => {});
+      await reconcileDatabaseWork(owner, topic);
+      expect(databaseWorkPending(owner, topic)).toBe(false);
+      expect(await db.execute(sql`select * from work_signal_test where id = ${topic}`)).toHaveLength(outcome === "commit" ? 1 : 0);
+    } finally { release(); await serverDone?.catch(() => {}); }
+  });
+
+  it("recovers after a real backend disconnect without replaying the queue write", async () => {
+    await expect(db.transaction(async tx => {
+      await signalDatabaseWork(tx, "disconnect");
+      await tx.execute(sql`insert into work_signal_test values ('disconnect')`);
+      const [{ pid }] = await tx.execute(sql`select pg_backend_pid() as pid`);
+      await db.execute(sql`select pg_terminate_backend(${pid as number})`);
+      await tx.execute(sql`select 1`);
+    })).rejects.toThrow();
+    await reconcileDatabaseWork(db, "disconnect");
+    expect(databaseWorkPending(db, "disconnect")).toBe(false);
+    expect(await db.execute(sql`select * from work_signal_test where id = 'disconnect'`)).toHaveLength(0);
+  });
+
+  it("shares notifications with dedicated child pools but isolates unrelated roots and topics", async () => {
+    const listener = vi.fn();
+    const other = createDb(temporary.connectionString);
+    const unsubscribe = subscribeDatabaseWork(db, "dedicated", listener);
+    try {
+      await other.transaction(tx => signalDatabaseWork(tx, "dedicated"));
+      await db.transaction(tx => signalDatabaseWork(tx, "other"));
+      expect(listener).not.toHaveBeenCalled();
+      await withDedicatedDbConnection(db, dedicated => dedicated.transaction(tx => signalDatabaseWork(tx, "dedicated")));
+      expect(listener.mock.calls.map(([event]) => event)).toEqual(["intent", "settled"]);
+    } finally { unsubscribe(); await other.$client.end(); }
+  });
+
+  it("never changes transaction success because a listener throws", async () => {
+    const warn = vi.spyOn(process, "emitWarning").mockImplementation(() => {});
+    const unsubscribe = subscribeDatabaseWork(db, "listener", () => { throw new Error("listener"); });
+    try {
+      await expect(db.transaction(async tx => { await signalDatabaseWork(tx, "listener"); return 7; })).resolves.toBe(7);
+      expect(warn).toHaveBeenCalledTimes(2);
+    } finally { unsubscribe(); warn.mockRestore(); }
+  });
+});
+
+describe("work signal query boundaries", () => {
+  it("adds no SQL to transactions which do not register queue work", async () => {
+    const execute = vi.fn();
+    const connection = installDatabaseWorkSignals({ execute, transaction: async (callback: any) => callback({ execute, transaction: async () => {} }) });
+    await expect(connection.transaction(async () => 19)).resolves.toBe(19);
+    expect(execute).not.toHaveBeenCalled();
+  });
+  it("fails before the write if XID acquisition fails, without retaining uncertainty", async () => {
+    const execute = vi.fn().mockRejectedValue(new Error("disconnected"));
+    const connection = installDatabaseWorkSignals({ execute, transaction: async (callback: any) => callback({ execute, transaction: async () => {} }) });
+    const write = vi.fn();
+    await expect(connection.transaction(async (tx: any) => { await signalDatabaseWork(tx, "failed"); write(); })).rejects.toThrow("disconnected");
+    expect(write).not.toHaveBeenCalled();
+    expect(databaseWorkPending(connection, "failed")).toBe(false);
+  });
+  it("rejects registration on an autocommit root", async () => {
+    const connection = installDatabaseWorkSignals({ execute: vi.fn(), transaction: vi.fn() });
+    await expect(signalDatabaseWork(connection, "bad")).rejects.toThrow("inside a transaction");
+  });
+});

packages/db/src/work-signals.ts

@@ -0,0 +1,109 @@
+import { sql } from "drizzle-orm";
+
+// Process-local hints about existing durable queues. No SQL triggers, tables,
+// background timers, or driver query replay. Only opted-in transactions fetch
+// an XID, so unrelated transactions keep their existing SQL behavior.
+type Listener = (event: "intent" | "settled") => void;
+type Transaction = { topics: Set<string>; xid?: Promise<string>; uncertain: boolean };
+type Scope = { listeners: Map<string, Set<Listener>>; transactions: Set<Transaction>; root: Connection };
+type Context = { scope: Scope; transaction?: Transaction };
+type Connection = { transaction: (...args: any[]) => Promise<any>; execute: (...args: any[]) => Promise<any> };
+const contexts = new WeakMap<object, Context>();
+
+function publish(scope: Scope, topic: string, event: "intent" | "settled" = "settled") {
+  for (const listener of scope.listeners.get(topic) ?? []) {
+    try { listener(event); }
+    catch (error) { process.emitWarning(`Database work listener failed: ${String(error)}`); }
+  }
+}
+
+/** Installed once by createDb; dedicated child pools share their owner's scope. */
+export function installDatabaseWorkSignals<T extends Connection>(db: T, owner?: object): T {
+  const scope = (owner && contexts.get(owner)?.scope) || { listeners: new Map(), transactions: new Set(), root: db };
+  install(db, { scope });
+  return db;
+}
+
+function install(connection: Connection, context: Context) {
+  contexts.set(connection, context);
+  const transaction = connection.transaction;
+  connection.transaction = async function (callback, ...options) {
+    // Nested savepoints belong to the outermost transaction. Even a released
+    // savepoint must not wake a consumer before its parent commits.
+    const state = context.transaction ?? { topics: new Set<string>(), uncertain: false };
+    try {
+      const result = await transaction.call(this, async (tx: Connection) => {
+        install(tx, { scope: context.scope, transaction: state });
+        return callback(tx);
+      }, ...options);
+      if (!context.transaction) context.scope.transactions.delete(state);
+      return result;
+    } catch (error) {
+      if (!context.transaction) {
+        // An XID is known before the first opted-in write can execute. If
+        // acquisition itself failed there cannot be a corresponding write.
+        if (state.xid) state.uncertain = true;
+        else context.scope.transactions.delete(state);
+      }
+      throw error;
+    } finally {
+      if (!context.transaction) for (const topic of state.topics) publish(context.scope, topic);
+    }
+  };
+}
+
+/** Await BEFORE the queue write, on its transaction. Nested transactions work. */
+export async function signalDatabaseWork(connection: object, topic: string): Promise<void> {
+  const context = contexts.get(connection);
+  if (!context) return; // Lightweight service test doubles have no DB lifecycle.
+  const state = context.transaction;
+  if (!state) throw new Error("Database work must be registered inside a transaction");
+  state.topics.add(topic);
+  context.scope.transactions.add(state);
+  publish(context.scope, topic, "intent");
+  if (!state.xid) {
+    state.xid = (connection as Connection).execute(sql`select pg_current_xact_id()::text as xid`)
+      .then((rows: { xid: string }[]) => rows[0]!.xid);
+  }
+  // Keep a failed acquisition distinguishable from an uncertain queue write.
+  try { await state.xid; }
+  catch (error) { state.xid = undefined; throw error; }
+}
+
+export function databaseWorkPending(owner: object, topic: string): boolean {
+  return [...(contexts.get(owner)?.scope.transactions ?? [])].some(state => state.topics.has(topic));
+}
+
+/** Resolve only failed transactions, on a fresh root-pool statement. Call
+ * BEFORE the queue scan so rows from a newly confirmed commit are visible.
+ * NULL means the XID aged out; it can no longer be in progress, so scanning
+ * the durable queue is sufficient even though its outcome is unavailable.
+ */
+export async function reconcileDatabaseWork(owner: object, topic: string): Promise<void> {
+  const scope = contexts.get(owner)?.scope;
+  if (!scope) return;
+  for (const state of [...scope.transactions]) {
+    if (!state.uncertain || !state.topics.has(topic) || !state.xid) continue;
+    const xid = await state.xid;
+    const rows = await scope.root.execute(sql`select pg_xact_status(${xid}::xid8) as status`);
+    if (rows[0]?.status === "in progress") continue;
+    if (!rows.length || ![null, "committed", "aborted"].includes(rows[0].status)) {
+      throw new Error("Unexpected database transaction status");
+    }
+    scope.transactions.delete(state);
+    for (const affected of state.topics) publish(scope, affected);
+  }
+}
+
+/** Includes intent, outer settlement, and uncertainty-resolution changes. */
+export function subscribeDatabaseWork(owner: object, topic: string, listener: Listener): () => void {
+  const context = contexts.get(owner);
+  if (!context) throw new Error("Database work subscriptions require createDb");
+  let listeners = context.scope.listeners.get(topic);
+  if (!listeners) context.scope.listeners.set(topic, listeners = new Set());
+  listeners.add(listener);
+  if (databaseWorkPending(owner, topic)) {
+    try { listener("intent"); } catch (error) { process.emitWarning(`Database work listener failed: ${String(error)}`); }
+  }
+  return () => { listeners.delete(listener); };
+}

server/src/__tests__/chat-completion-delivery.test.ts

@@ -1,3 +1,5 @@
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { subscribeDeliveryWork, DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, asc, eq } from "drizzle-orm";
 import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -7,7 +9,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
 import { settleInterruptedNativeBootstrap, terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
 import { getExecutionBlocker } from "../services/execution-blocker.js";
 import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js";
-import { issueService } from "../services/issues.js";
+import { executeIssuePostCommitActions, type IssuePostCommitAction, issueService } from "../services/issues.js";
 import { buildLowTrustSourceTrust } from "../services/source-trust.js";
 import { documentService } from "../services/documents.js";
 import { instanceSettingsService } from "../services/instance-settings.js";
@@ -51,6 +53,57 @@ const support = await getEmbeddedPostgresTestSupport();
     };
     return { companyId, agentId, sourceId, runId, task, create, finish, rows, due, wakeup, service, run };
   }
+  it("dispatches a committed completion without activity publication and recovers after restart", async () => {
+    const f = await seed();
+    const run = vi.fn(() => f.service.sweepPending());
+    const makeWorker = () => {
+      const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
+      const worker = coordinator.register(DELIVERY_QUEUES.chatCompletion, { retryMs: 60_000, run, hasPending: f.service.hasPending });
+      return { ...worker, stop: coordinator.stop };
+    };
+    const worker = makeWorker();
+    try {
+      await worker.ready;
+      expect(await f.service.hasPending()).toBe(false);
+      expect(run).toHaveBeenCalledTimes(1);
+      await f.finish(); // Service mutation alone: no route activity callback.
+      await vi.waitFor(() => expect(f.wakeup).toHaveBeenCalledTimes(1));
+      expect(await f.service.hasPending()).toBe(true); // queued receipt still owed
+    } finally { await worker.stop(); }
+    // Admit the first reporting turn; later completions need a new turn.
+    await f.run();
+    // A second task commits while no consumer is registered.
+    const next = await f.create();
+    await f.finish(next.id);
+    const restarted = makeWorker();
+    try {
+      await restarted.ready;
+      expect(f.wakeup).toHaveBeenCalledTimes(2);
+    } finally { await restarted.stop(); }
+  });
+  it("automatically notifies after a caller-owned transaction commits", async () => {
+    const f = await seed();
+    const notified = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatCompletion, notified);
+    try {
+      const actions: IssuePostCommitAction[] = [];
+      await db.transaction(async tx => {
+        await issueService(db).update(f.task.id, { status: "done" }, tx, [], actions);
+        expect(notified).not.toHaveBeenCalled();
+      });
+      expect(notified).toHaveBeenCalledTimes(1);
+      expect(actions).toEqual([]);
+      await executeIssuePostCommitActions(db, actions);
+      expect(notified).toHaveBeenCalledTimes(1);
+      const rolledBack: IssuePostCommitAction[] = [];
+      await expect(db.transaction(async tx => {
+        await issueService(db).update(f.task.id, { status: "todo" }, tx, [], rolledBack);
+        throw new Error("rollback");
+      })).rejects.toThrow("rollback");
+      expect(notified).toHaveBeenCalledTimes(1);
+      expect((await f.rows())[0]?.status).toBe("pending");
+    } finally { unsubscribe(); }
+  });
   it("records authenticated origins and atomically creates one event per Done transition", async () => {
     const f = await seed();
     expect(await db.select().from(handoffs).where(eq(handoffs.taskId, f.task.id))).toMatchObject([{ conversationId: f.sourceId, sessionGeneration: 0 }]);

server/src/__tests__/connection-intents-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
@@ -773,7 +775,11 @@ describeEmbeddedPostgres("connectionIntentService", () => {
   it("recovers a resolution after a failed dispatch and acknowledges an already queued wake exactly once", async () => {
     const service = connectionIntentService(db);
     const pending = await service.request(claims, "airtable");
+    const notified = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.connection, notified);
     await service.decline(pending.interactionId!, claims.responsible_user_id);
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
     const wakeup = vi.fn().mockRejectedValueOnce(new Error("simulated crash before durable enqueue"));
     const deliveries = connectionIntentDeliveryService(db, { wakeup } as never);
     await expect(deliveries.deliver(pending.interactionId!)).rejects.toThrow("simulated crash");

server/src/__tests__/delivery-work-coordinator.test.ts

@@ -0,0 +1,192 @@
+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 { notifyDeliveryWork, DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
+afterEach(async () => { for (const c of coordinators.splice(0)) await c.stop(); vi.useRealTimers(); });
+function setup() {
+  vi.useFakeTimers();
+  let outcome = "committed";
+  let rejectCommit = false;
+  const execute = vi.fn(async () => [{ xid: "42", status: outcome }]);
+  const owner = installDatabaseWorkSignals({ execute, async transaction(callback: (tx: any) => Promise<any>) {
+    const result = await callback({ execute, transaction: owner.transaction });
+    if (rejectCommit) throw new Error("lost commit reply");
+    return result;
+  } });
+  const canRun = vi.fn(() => true), canReconcile = vi.fn(() => false), onError = vi.fn();
+  const coordinator = createDeliveryWorkCoordinator({ owner, canRun, canReconcile, onError });
+  coordinators.push(coordinator);
+  const enqueue = (queue = DELIVERY_QUEUES.feedback as typeof DELIVERY_QUEUES[keyof typeof DELIVERY_QUEUES]) =>
+    owner.transaction(tx => notifyDeliveryWork(tx, queue));
+  return { owner, canRun, canReconcile, onError, coordinator, enqueue, execute,
+    failCommit: () => { rejectCommit = true; outcome = "in progress"; }, settle: (status = "committed") => { outcome = status; } };
+}
+function task() { return { retryMs: 5000, run: vi.fn(async () => {}), hasPending: vi.fn(async () => false) }; }
+
+describe("delivery work coordinator", () => {
+  it("leaves five empty queues with no deadlines, timers, or periodic queries", async () => {
+    const s = setup();
+    const tasks = Object.values(DELIVERY_QUEUES).map(queue => {
+      const t = task(); return { queue, t, worker: s.coordinator.register(queue, t) };
+    });
+    await Promise.all(tasks.map(t => t.worker.ready));
+    expect(vi.getTimerCount()).toBe(0);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+    await vi.advanceTimersByTimeAsync(3_600_000);
+    for (const { queue, t } of tasks) {
+      expect(t.run).toHaveBeenCalledTimes(1);
+      await s.enqueue(queue); await s.enqueue(queue);
+    }
+    await vi.advanceTimersByTimeAsync(1);
+    for (const { t } of tasks) expect(t.run).toHaveBeenCalledTimes(2);
+    expect(vi.getTimerCount()).toBe(0);
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("does not postpone an already committed wake when another producer starts writing", async () => {
+    const s = setup(), t = task();
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;
+    await s.enqueue();
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const writing = s.owner.transaction(async tx => {
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.feedback);
+      await gate;
+    });
+    await vi.advanceTimersByTimeAsync(1);
+    expect(t.run).toHaveBeenCalledTimes(2);
+    release(); await writing;
+    await vi.advanceTimersByTimeAsync(1);
+    expect(t.run).toHaveBeenCalledTimes(3);
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("fences unresolved writes until the server settles, then scans and releases the idle hold", async () => {
+    const s = setup(), t = task();
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;
+    s.failCommit();
+    await expect(s.enqueue()).rejects.toThrow("lost commit reply");
+    expect(idleWorkSnapshot().active).toBe(1);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(t.run).toHaveBeenCalledTimes(2); // Empty scan cannot clear unresolved write.
+    expect(idleWorkSnapshot().active).toBe(1);
+    expect(s.coordinator.nextWakeAt()).not.toBeNull();
+    s.settle();
+    await vi.advanceTimersByTimeAsync(5001);
+    expect(t.run.mock.calls.length).toBeGreaterThanOrEqual(3);
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("retains unresolved intent across coordinator replacement", async () => {
+    const s = setup(), t = task();
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;
+    s.failCommit(); await expect(s.enqueue()).rejects.toThrow();
+    await s.coordinator.stop();
+    expect(idleWorkSnapshot().active).toBe(0);
+    const replacement = createDeliveryWorkCoordinator({ owner: s.owner, canRun: () => true, onError: vi.fn() });
+    coordinators.push(replacement);
+    await replacement.register(DELIVERY_QUEUES.feedback, t).ready;
+    expect(idleWorkSnapshot().active).toBe(1);
+    s.settle(); await vi.advanceTimersByTimeAsync(5001);
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(replacement.nextWakeAt()).toBeNull();
+  });
+  it("retries pending deliveries and transient failures, then goes quiet", async () => {
+    const s = setup(), t = task();
+    t.run.mockRejectedValueOnce(new Error("database disconnected"));
+    t.hasPending.mockResolvedValueOnce(true);
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;
+    expect(s.onError).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(t.run).toHaveBeenCalledTimes(3);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("does not overlap a sweep or lose a commit during its final pending check", async () => {
+    const s = setup(), t = task();
+    let release!: (pending: boolean) => void;
+    t.hasPending.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
+    s.coordinator.register(DELIVERY_QUEUES.connection, t);
+    await vi.advanceTimersByTimeAsync(1);
+    await s.enqueue(DELIVERY_QUEUES.connection);
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(t.run).toHaveBeenCalledTimes(1);
+    release(false);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(t.run).toHaveBeenCalledTimes(2);
+  });
+  it("reconciles a rollback during idle drain and releases the empty-queue hold without dispatching", async () => {
+    const s = setup(), t = task();
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;
+    s.canRun.mockReturnValue(false);
+    s.canReconcile.mockReturnValue(true);
+    s.failCommit(); await expect(s.enqueue()).rejects.toThrow("lost commit reply");
+    await vi.advanceTimersByTimeAsync(1);
+    expect(idleWorkSnapshot().active).toBe(1);
+    expect(t.run).toHaveBeenCalledTimes(1);
+    s.settle("aborted");
+    await vi.advanceTimersByTimeAsync(5001);
+    expect(t.run).toHaveBeenCalledTimes(1);
+    expect(t.hasPending).toHaveBeenCalled();
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(s.coordinator.nextWakeAt()).toBeNull();
+  });
+  it("keeps committed pending work recoverable during drain without starting deliveries", async () => {
+    const s = setup(), t = task();
+    await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready;
+    s.canRun.mockReturnValue(false);
+    s.canReconcile.mockReturnValue(true);
+    t.hasPending.mockResolvedValue(true);
+    await s.enqueue();
+    await vi.advanceTimersByTimeAsync(1);
+    expect(t.run).toHaveBeenCalledTimes(1);
+    expect(s.coordinator.nextWakeAt()).not.toBeNull();
+    s.canRun.mockReturnValue(true);
+    t.hasPending.mockResolvedValue(false);
+    await vi.advanceTimersByTimeAsync(5001);
+    expect(t.run).toHaveBeenCalledTimes(2);
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("does not query during standby and resumes when admission opens", async () => {
+    const s = setup(), t = task(); s.canRun.mockReturnValue(false);
+    await s.coordinator.register(DELIVERY_QUEUES.toolAction, t).ready;
+    await s.enqueue(DELIVERY_QUEUES.toolAction);
+    await vi.advanceTimersByTimeAsync(120_000);
+    expect(t.run).not.toHaveBeenCalled();
+    expect(t.hasPending).not.toHaveBeenCalled();
+    s.canRun.mockReturnValue(true);
+    await vi.advanceTimersByTimeAsync(5001);
+    expect(t.run).toHaveBeenCalledTimes(1);
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("awaits in-flight work on shutdown and releases the subscription for restart", async () => {
+    const s = setup(), t = task();
+    let release!: () => void;
+    t.run.mockImplementationOnce(() => new Promise(resolve => { release = resolve; }));
+    s.coordinator.register(DELIVERY_QUEUES.chatCompletion, t);
+    await vi.advanceTimersByTimeAsync(1);
+    const stopped = vi.fn();
+    const stopping = s.coordinator.stop().then(stopped);
+    await s.enqueue(DELIVERY_QUEUES.chatCompletion);
+    expect(stopped).not.toHaveBeenCalled();
+    release(); await stopping;
+    expect(vi.getTimerCount()).toBe(0);
+    expect(idleWorkSnapshot().active).toBe(0);
+  });
+  it("cancels an abortable delivery before awaiting shutdown", async () => {
+    const s = setup();
+    const hasPending = vi.fn(async () => false);
+    const run = vi.fn((signal: AbortSignal) => new Promise<void>((_resolve, reject) => {
+      signal.addEventListener("abort", () => reject(signal.reason), { once: true });
+    }));
+    s.coordinator.register(DELIVERY_QUEUES.feedback, { retryMs: 5000, run, hasPending });
+    await vi.advanceTimersByTimeAsync(1);
+    await s.coordinator.stop();
+    expect(run.mock.calls[0]?.[0].aborted).toBe(true);
+    expect(hasPending).not.toHaveBeenCalled();
+    expect(s.onError).not.toHaveBeenCalled();
+    expect(idleWorkSnapshot().active).toBe(0);
+    expect(vi.getTimerCount()).toBe(0);
+  });
+});

server/src/__tests__/feedback-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import fs from "node:fs";
 import os from "node:os";
@@ -800,6 +802,8 @@ describeEmbeddedPostgres("feedbackService.saveIssueVote", () => {
       sessionId,
     });
 
+    const notified = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.feedback, notified);
     await flushingSvc.saveIssueVote({
       issueId,
       targetType: "issue_comment",
@@ -808,7 +812,12 @@ describeEmbeddedPostgres("feedbackService.saveIssueVote", () => {
       authorUserId: "user-1",
       allowSharing: true,
     });
-    await flushingSvc.flushPendingFeedbackTraces();
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
+    expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(true);
+    // The route and worker may request a flush together after the commit.
+    await Promise.all([flushingSvc.flushPendingFeedbackTraces(), flushingSvc.flushPendingFeedbackTraces()]);
+    expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(false);
 
     expect(uploadTraceBundle).toHaveBeenCalledTimes(1);
     const bundle = uploadTraceBundle.mock.calls[0]?.[0] as Record<string, unknown> | undefined;
@@ -1152,6 +1161,30 @@ describeEmbeddedPostgres("feedbackService.saveIssueVote", () => {
     expect(uploadTraceBundle).toHaveBeenCalledTimes(1);
   });
 
+  it("leaves aborted and unstarted exports available for recovery after shutdown", async () => {
+    const first = await seedIssueWithAgentComment();
+    const second = await seedIssueWithAgentComment();
+    const shutdown = new AbortController();
+    const uploadTraceBundle = vi.fn().mockResolvedValue({ objectKey: "saved" })
+      .mockImplementationOnce(async (_bundle, signal: AbortSignal) => {
+        shutdown.abort(new Error("shutdown"));
+        signal.throwIfAborted();
+      });
+    const flushingSvc = feedbackService(db, { shareClient: { uploadTraceBundle } });
+    for (const item of [first, second]) {
+      await flushingSvc.saveIssueVote({ issueId: item.issueId, targetType: "issue_comment",
+        targetId: item.commentId, vote: "up", authorUserId: "user-1", allowSharing: true });
+    }
+    expect(await flushingSvc.flushPendingFeedbackTraces({ signal: shutdown.signal }))
+      .toMatchObject({ attempted: 1, sent: 0, failed: 1 });
+    const rows = await db.select().from(feedbackExports);
+    expect(rows.filter(row => row.status === "failed")).toHaveLength(1);
+    expect(rows.filter(row => row.status === "pending")).toHaveLength(1);
+    expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(true);
+    expect(await flushingSvc.flushPendingFeedbackTraces()).toMatchObject({ attempted: 2, sent: 2, failed: 0 });
+    expect(await flushingSvc.hasPendingFeedbackTraces()).toBe(false);
+  });
+
   it("marks pending shared traces as failed when no feedback export backend is configured", async () => {
     const { companyId, issueId, commentId } = await seedIssueWithAgentComment();
 

server/src/__tests__/feedback-share-client.test.ts

@@ -1,6 +1,6 @@
 import { gunzipSync } from "node:zlib";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
-import { createFeedbackTraceShareClientFromConfig } from "../services/feedback-share-client.js";
+import { FEEDBACK_UPLOAD_TIMEOUT_MS, createFeedbackTraceShareClientFromConfig } from "../services/feedback-share-client.js";
 
 describe("feedback trace share client", () => {
   beforeEach(() => {
@@ -12,6 +12,7 @@ describe("feedback trace share client", () => {
 
   afterEach(() => {
     vi.restoreAllMocks();
+    vi.unstubAllGlobals();
   });
 
   it("defaults to telemetry.paperclip.ing when no backend url is configured", async () => {
@@ -98,4 +99,43 @@ describe("feedback trace share client", () => {
     expect(parsed.objectKey.endsWith("/export-1.json")).toBe(true);
     expect(parsed.bundle.envelope).toEqual({ hello: "world" });
   });
+  it.each(["headers", "body"])("cancels a stalled upload at its deadline while waiting for %s", async phase => {
+    const deadline = new AbortController();
+    const timeout = vi.spyOn(AbortSignal, "timeout").mockReturnValue(deadline.signal);
+    let started!: () => void;
+    const waiting = new Promise<void>(resolve => { started = resolve; });
+    vi.mocked(fetch).mockImplementation(async (_url, options) => {
+      const block = () => new Promise<never>((_resolve, reject) => {
+        options!.signal!.addEventListener("abort", () => reject(options!.signal!.reason), { once: true });
+        started();
+      });
+      if (phase === "headers") return block();
+      return { ok: true, json: block } as unknown as Response;
+    });
+    const client = createFeedbackTraceShareClientFromConfig({ feedbackExportBackendUrl: undefined, feedbackExportBackendToken: undefined });
+    const bundle = { companyId: "company", traceId: "trace" } as Parameters<typeof client.uploadTraceBundle>[0];
+    const uploading = client.uploadTraceBundle(bundle);
+    const rejected = expect(uploading).rejects.toThrow("upload deadline");
+    await waiting;
+    expect(timeout).toHaveBeenCalledWith(FEEDBACK_UPLOAD_TIMEOUT_MS);
+    deadline.abort(new Error("upload deadline"));
+    await rejected;
+  });
+
+  it("passes shutdown cancellation through to a pending upload", async () => {
+    const shutdown = new AbortController();
+    let started!: () => void;
+    const waiting = new Promise<void>(resolve => { started = resolve; });
+    vi.mocked(fetch).mockImplementation((_url, options) => new Promise((_resolve, reject) => {
+      options!.signal!.addEventListener("abort", () => reject(options!.signal!.reason), { once: true });
+      started();
+    }));
+    const client = createFeedbackTraceShareClientFromConfig({ feedbackExportBackendUrl: undefined, feedbackExportBackendToken: undefined });
+    const bundle = { companyId: "company", traceId: "trace" } as Parameters<typeof client.uploadTraceBundle>[0];
+    const rejected = expect(client.uploadTraceBundle(bundle, shutdown.signal)).rejects.toThrow("shutdown");
+    await waiting;
+    shutdown.abort(new Error("shutdown"));
+    await rejected;
+  });
+
 });

server/src/__tests__/issue-feedback-routes.test.ts

@@ -8,6 +8,7 @@ const mockFeedbackService = vi.hoisted(() => ({
   listIssueVotesForUser: vi.fn(),
   listFeedbackTraces: vi.fn(),
   saveIssueVote: vi.fn(),
+  flushPendingFeedbackTraces: vi.fn(() => new Promise(() => {})),
 }));
 
 const mockIssueService = vi.hoisted(() => ({
@@ -18,9 +19,6 @@ const mockIssueService = vi.hoisted(() => ({
   findMentionedAgents: vi.fn(),
 }));
 
-const mockFeedbackExportService = vi.hoisted(() => ({
-  flushPendingFeedbackTraces: vi.fn(async () => ({ attempted: 1, sent: 1, failed: 0 })),
-}));
 const mockAccessService = vi.hoisted(() => ({
   canUser: vi.fn(),
   hasPermission: vi.fn(),
@@ -138,7 +136,7 @@ async function createApp(actor: Record<string, unknown>) {
     (req as any).actor = actor;
     next();
   });
-  app.use("/api", issueRoutes({} as any, {} as any, { feedbackExportService: mockFeedbackExportService }));
+  app.use("/api", issueRoutes({} as any, {} as any));
   const routeErrors: string[] = [];
   app.locals.routeErrors = routeErrors;
   app.use((error: unknown, _req: express.Request, _res: express.Response, next: express.NextFunction) => {
@@ -152,11 +150,6 @@ async function createApp(actor: Record<string, unknown>) {
 describe("issue feedback trace routes", () => {
   beforeEach(() => {
     vi.clearAllMocks();
-    mockFeedbackExportService.flushPendingFeedbackTraces.mockResolvedValue({
-      attempted: 1,
-      sent: 1,
-      failed: 0,
-    });
     mockHeartbeatService.wakeup.mockResolvedValue(undefined);
     mockHeartbeatService.reportRunActivity.mockResolvedValue(undefined);
     mockHeartbeatService.getRun.mockResolvedValue(null);
@@ -174,7 +167,7 @@ describe("issue feedback trace routes", () => {
     mockLogActivity.mockResolvedValue(undefined);
   });
 
-  it("flushes a newly shared feedback trace immediately after saving the vote", async () => {
+  it("returns the saved vote without waiting for feedback uploads", async () => {
     const targetId = "11111111-1111-4111-8111-111111111111";
     mockIssueService.getById.mockResolvedValue({
       id: "issue-1",
@@ -211,11 +204,9 @@ describe("issue feedback trace routes", () => {
       });
 
     expect([200, 201]).toContain(res.status);
-    expect(mockFeedbackExportService.flushPendingFeedbackTraces).toHaveBeenCalledWith({
-      companyId: "company-1",
-      traceId: "trace-1",
-      limit: 1,
-    });
+    expect(mockFeedbackService.saveIssueVote).toHaveBeenCalledTimes(1);
+    expect(mockFeedbackService.flushPendingFeedbackTraces).not.toHaveBeenCalled();
+
   });
 
   it("rejects non-board callers before fetching a feedback trace", async () => {

server/src/__tests__/issue-thread-interactions-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
@@ -231,11 +233,15 @@ describeEmbeddedPostgres("issueThreadInteractionService", () => {
     expect(created.payload.questions.map((question) => question.id)).toEqual(["repo", "scope", "hosting"]);
     expect(await interactionsSvc.create(issue, input, { userId: "local-board" })).toEqual(created);
     await expect(interactionsSvc.answerQuestions(issue, created.id, { answers: [{ questionId: "repo", optionIds: [], otherText: "https://example.com/repo" }] }, { userId: "local-board" })).rejects.toThrow("requires an answer");
+    const notified = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.question, notified);
     const answered = await interactionsSvc.answerQuestions(issue, created.id, { answers: [
       { questionId: "repo", optionIds: [], otherText: "https://example.com/repo" },
       { questionId: "scope", optionIds: ["all"] },
       { questionId: "hosting", optionIds: ["existing", "new"] },
     ] }, { userId: "local-board" });
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
     expect(answered.status).toBe("answered");
   });
 

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -44,11 +44,17 @@ const {
   const completionSweepMock = vi.fn(async () => undefined);
   const createAppMock = vi.fn(async () => Object.assign((_: unknown, __: unknown) => {}, {
     locals: {
+      deliveryWork: {
+        register: (_queue: string, task: { run: () => Promise<unknown> }) => ({
+          ready: task.run().catch(() => undefined), wake: vi.fn(),
+        }),
+        stop: vi.fn(async () => {}),
+      },
       toolGateway: {
         sweepActionReviews: vi.fn(async () => ({ scanned: 0 })),
         cleanupExpiredSessions: vi.fn(async () => ({ deletedCount: 0 })),
       },
-      toolActionDeliveries: { sweepPending: vi.fn(async () => ({ scanned: 0, delivered: 0 })) },
+      toolActionDeliveries: { hasPending: vi.fn(async () => false), sweepPending: vi.fn(async () => ({ scanned: 0, delivered: 0 })) },
     },
   }) as never);
   const createBetterAuthInstanceMock = vi.fn(() => ({}));
@@ -133,6 +139,7 @@ const {
   };
   const routineServiceFactoryMock = vi.fn(() => routineServiceMock);
   const feedbackExportServiceMock = {
+    hasPendingFeedbackTraces: vi.fn(async () => false),
     flushPendingFeedbackTraces: vi.fn(async () => ({ attempted: 0, sent: 0, failed: 0 })),
   };
   const feedbackServiceFactoryMock = vi.fn(() => feedbackExportServiceMock);
@@ -363,16 +370,18 @@ vi.mock("../services/index.js", () => ({
   })),
 }));
 
-vi.mock("../services/chat-completion-delivery.js", () => ({ chatCompletionDeliveryService: () => ({ sweepPending: completionSweepMock }) }));
+vi.mock("../services/chat-completion-delivery.js", () => ({ chatCompletionDeliveryService: () => ({ hasPending: vi.fn(async () => false), sweepPending: completionSweepMock }) }));
 
 vi.mock("../services/connection-intent-delivery.js", () => ({
   connectionIntentDeliveryService: vi.fn(() => ({
+    hasPending: vi.fn(async () => false),
     sweepPending: vi.fn(async () => ({ scanned: 0, failed: 0 })),
   })),
 }));
 
 vi.mock("../services/question-response-delivery.js", () => ({
   questionResponseDeliveryService: vi.fn(() => ({
+    hasPending: vi.fn(async () => false),
     sweepPending: vi.fn(async () => ({
       scanned: 0,
       steered: 0,
@@ -675,6 +684,22 @@ describe("startServer feedback export wiring", () => {
     expect(completionSweepMock).toHaveBeenCalled();
   });
 
+  it("keeps activity-driven completion scans scoped to the changed task", async () => {
+    await startServer();
+    completionSweepMock.mockClear();
+    const { publishLiveEvent } = await import("../services/live-events.js");
+    publishLiveEvent({ companyId: "company-one", type: "activity.logged", payload: {
+      action: "issue.updated", entityId: "task-one",
+    } });
+    expect(completionSweepMock).toHaveBeenCalled();
+    for (const args of completionSweepMock.mock.calls) {
+      expect(args).toEqual([{ companyId: "company-one", taskId: "task-one" }]);
+    }
+    completionSweepMock.mockClear();
+    publishLiveEvent({ companyId: "company-two", type: "activity.logged", payload: { action: "issue.updated" } });
+    expect(completionSweepMock).not.toHaveBeenCalled();
+  });
+
   it("never invokes the retired review detector at startup or on periodic recovery", async () => {
     loadConfigMock.mockReturnValue(buildTestConfig({
       heartbeatSchedulerEnabled: true,

server/src/__tests__/tool-gateway-service.test.ts

@@ -1,3 +1,5 @@
+import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
+import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
 import { randomUUID } from "node:crypto";
 import { and, eq } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
@@ -219,6 +221,8 @@ describeEmbeddedPostgres("tool gateway service", () => {
       runId: run.id,
     });
 
+    const notified = vi.fn();
+    const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.toolAction, notified);
     await expect(gateway.executeTool({
       sessionToken: session.token,
       tool: "mcp-remote-fixture:update_note",
@@ -228,6 +232,9 @@ describeEmbeddedPostgres("tool gateway service", () => {
       details: { instructions: expect.stringContaining("A human approval card was posted on task") },
     });
 
+    unsubscribe();
+    expect(notified).toHaveBeenCalledTimes(1);
+
     await expect(gateway.executeTool({
       sessionToken: session.token,
       tool: "mcp-remote-fixture:update_note",

server/src/__tests__/work-scheduler.test.ts

@@ -0,0 +1,47 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { createWorkScheduler } from "../services/work-scheduler.js";
+afterEach(() => vi.useRealTimers());
+describe("work scheduler", () => {
+  it("owns one timer, replaces named deadlines, and exposes the earliest wake", async () => {
+    vi.useFakeTimers(); vi.setSystemTime(0);
+    const s = createWorkScheduler();
+    const first = vi.fn(), second = vi.fn();
+    s.schedule("first", 100, first);
+    s.schedule("second", 50, second);
+    s.schedule("first", 20, first);
+    expect(vi.getTimerCount()).toBe(1);
+    expect(s.nextWakeAt()).toBe(20);
+    await vi.advanceTimersByTimeAsync(20);
+    expect(first).toHaveBeenCalledTimes(1);
+    expect(second).not.toHaveBeenCalled();
+    expect(s.nextWakeAt()).toBe(50);
+    s.cancel("second");
+    expect(s.nextWakeAt()).toBeNull();
+    expect(vi.getTimerCount()).toBe(0);
+  });
+  it("honors cancellation during dispatch and isolates a failing callback", async () => {
+    vi.useFakeTimers(); vi.setSystemTime(0);
+    const warn = vi.spyOn(process, "emitWarning").mockImplementation(() => {});
+    const s = createWorkScheduler(), cancelled = vi.fn(), later = vi.fn();
+    s.schedule("first", 10, () => { s.cancel("cancelled"); throw new Error("failure"); });
+    s.schedule("cancelled", 10, cancelled);
+    s.schedule("later", 20, later);
+    await vi.advanceTimersByTimeAsync(20);
+    expect(cancelled).not.toHaveBeenCalled();
+    expect(later).toHaveBeenCalledTimes(1);
+    expect(warn).toHaveBeenCalledTimes(1);
+    warn.mockRestore(); s.stop();
+  });
+  it("does not fire distant deadlines early and cannot restart after stop", async () => {
+    vi.useFakeTimers(); vi.setSystemTime(0);
+    const s = createWorkScheduler(), run = vi.fn();
+    s.schedule("far", 3_000_000_000, run);
+    await vi.advanceTimersByTimeAsync(2_147_483_647);
+    expect(run).not.toHaveBeenCalled();
+    await vi.advanceTimersByTimeAsync(852_516_353);
+    expect(run).toHaveBeenCalledTimes(1);
+    s.stop(); s.schedule("late", Date.now(), run);
+    expect(vi.getTimerCount()).toBe(0);
+    expect(() => createWorkScheduler().schedule("bad", NaN, run)).toThrow("finite");
+  });
+});

server/src/app.ts

@@ -1,3 +1,5 @@
+import { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./services/delivery-work-notifications.js";
 import { idleAdmissionMiddleware, trackIdleRequestHandlers } from "./middleware/idle-admission.js";
 import { isIdleTaskDrainActive, trackIdleWork } from "./services/task-admission.js";
 import { customerSuccessRoutes } from "./routes/customer-success.js";
@@ -483,11 +485,13 @@ export async function createApp(
     serverPort: number;
     storageService: StorageService;
     feedbackExportService?: {
+      hasPendingFeedbackTraces(): Promise<boolean>;
       flushPendingFeedbackTraces(input?: {
         companyId?: string;
         traceId?: string;
         limit?: number;
         now?: Date;
+        signal?: AbortSignal;
       }): Promise<unknown>;
     };
     databaseBackupService?: InstanceDatabaseBackupService;
@@ -877,7 +881,6 @@ export async function createApp(
   // route prefixes, so this dependency does not change issue-route precedence.
   api.use(issueRoutes(db, opts.storageService, {
     chatRunRetries: chatChannels,
-    feedbackExportService: opts.feedbackExportService,
     pluginWorkerManager: workerManager,
     approveToolActionRequest: (input) => toolGateway.approveActionRequest(input),
     declineToolActionRequest: (input) => toolGateway.declineActionRequest(input),
@@ -1190,40 +1193,19 @@ export async function createApp(
 
   jobCoordinator.start();
   scheduler.start();
-  let feedbackExportShuttingDown = false;
-  let feedbackExportTimer: ReturnType<typeof setInterval> | null = null;
-  const disableFeedbackExportFlushes = () => {
-    feedbackExportShuttingDown = true;
-    if (feedbackExportTimer) {
-      clearInterval(feedbackExportTimer);
-      feedbackExportTimer = null;
-    }
-  };
-  const flushPendingFeedbackExports = async () => {
-    if (feedbackExportShuttingDown || isWarmStandby() || isIdleTaskDrainActive()) return;
-    try {
-      await opts.feedbackExportService?.flushPendingFeedbackTraces();
-    } catch (err) {
-      if (isDatabaseConnectionUnavailableError(err)) {
-        disableFeedbackExportFlushes();
-        logger.warn(
-          { err },
-          "Disabling pending feedback export flushes because the database is unavailable",
-        );
-        return;
-      }
-      logger.error({ err }, "Failed to flush pending feedback exports");
-    }
-  };
-
-  feedbackExportTimer = opts.feedbackExportService
-    ? setInterval(() => {
-        void flushPendingFeedbackExports();
-      }, FEEDBACK_EXPORT_FLUSH_INTERVAL_MS)
-    : null;
-  feedbackExportTimer?.unref?.();
+  const deliveryWork = createDeliveryWorkCoordinator({
+    owner: db,
+    canRun: () => !isWarmStandby() && !isIdleTaskDrainActive(),
+    canReconcile: () => !isWarmStandby(),
+    onError: (err, queue) => logger.error({ err, queue }, "Delivery reconciliation failed"),
+  });
+  app.locals.deliveryWork = deliveryWork;
   if (opts.feedbackExportService) {
-    void flushPendingFeedbackExports();
+    deliveryWork.register(DELIVERY_QUEUES.feedback, {
+      retryMs: FEEDBACK_EXPORT_FLUSH_INTERVAL_MS,
+      run: (signal) => opts.feedbackExportService!.flushPendingFeedbackTraces({ signal }),
+      hasPending: () => opts.feedbackExportService!.hasPendingFeedbackTraces(),
+    });
   }
   emailChannels.start();
   const flushChatPublications = async () => {
@@ -1389,7 +1371,7 @@ export async function createApp(
       await publicMcpEvents?.stop();
       await dotMcpEvents?.stop();
       jobCoordinator.stop();
-      disableFeedbackExportFlushes();
+      await deliveryWork.stop();
       unsubscribeChatPublicationSignals();
       chatReconciliation.stop();
       if (chatPublicationTimer) {

server/src/index.ts

@@ -1,9 +1,11 @@
+import { subscribeAllCompanyLiveEvents } from "./services/live-events.js";
+import { DELIVERY_QUEUES } from "./services/delivery-work-notifications.js";
+import type { createDeliveryWorkCoordinator } from "./services/delivery-work-coordinator.js";
 import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js";
 import { markIdleStartupComplete } from "./services/idle-local-work.js";
 import { clearIdleBackupWakeRequired, idleBackupWakeMarker, markIdleBackupWakeRequired, verifyIdleDatabaseBackup } from "./services/idle-database-backup.js";
 import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js";
 import { createCloudWarmStandby } from "./services/cloud-warm-standby.js";
-import { subscribeAllCompanyLiveEvents } from "./services/live-events.js";
 import { chatCompletionDeliveryService } from "./services/chat-completion-delivery.js";
 /// <reference path="./types/express.d.ts" />
 // Kicks off the OTel bootstrap as early as possible (no-op unless
@@ -1252,20 +1254,31 @@ async function startServerWithDatabaseTeardown(
   const ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS = 5 * 60 * 1000;
   const environmentLeaseCleanupHeartbeat = accountingHeartbeat;
   const chatCompletionDeliveries = chatCompletionDeliveryService(db as any, environmentLeaseCleanupHeartbeat);
-  // Activity publication happens after the status transaction commits. This is
-  // a best-effort fast path; the durable outbox and sweeps remain authoritative.
-  const unsubscribeChatCompletions = subscribeAllCompanyLiveEvents(event => {
-    if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || event.type !== "activity.logged" ||
-      event.payload.action !== "issue.updated" || typeof event.payload.entityId !== "string") return;
-    trackHeartbeatSchedulerWork(chatCompletionDeliveries.sweepPending({ companyId: event.companyId, taskId: event.payload.entityId })
-      .catch(err => logger.error({ err }, "post-commit chat completion delivery failed")));
-  });
-  server.on("close", unsubscribeChatCompletions);
   const connectionDeliveries = connectionIntentDeliveryService(db as any, environmentLeaseCleanupHeartbeat);
   const questionResponseDeliveries = questionResponseDeliveryService(db as any, {
     heartbeat: environmentLeaseCleanupHeartbeat,
     resolveNativeQuestion: (interaction) => deliverNativeQuestionResponse(db as any, interaction),
   });
+  const deliveryWork = app.locals.deliveryWork as ReturnType<typeof createDeliveryWorkCoordinator>;
+  const deliveryWorkers = ([
+    [DELIVERY_QUEUES.chatCompletion, chatCompletionDeliveries],
+    [DELIVERY_QUEUES.connection, connectionDeliveries],
+    [DELIVERY_QUEUES.question, questionResponseDeliveries],
+    [DELIVERY_QUEUES.toolAction, app.locals.toolActionDeliveries as ReturnType<typeof import("./services/tool-action-delivery.js").toolActionDeliveryService>],
+  ] as const).map(([queue, service]) => deliveryWork.register(queue, {
+    retryMs: config.heartbeatSchedulerIntervalMs,
+    run: () => service.sweepPending(),
+    hasPending: () => service.hasPending(),
+  }));
+  // Preserve the existing task-scoped activity fast path. Claims in the
+  // delivery service arbitrate with the coordinator's startup/recovery sweep.
+  const unsubscribeChatCompletions = subscribeAllCompanyLiveEvents(event => {
+    if (isIdleTaskDrainActive() || isWarmStandby() || heartbeatSchedulerStopped || event.type !== "activity.logged" ||
+      event.payload.action !== "issue.updated" || typeof event.payload.entityId !== "string") return;
+    trackHeartbeatSchedulerWork(chatCompletionDeliveries.sweepPending({ companyId: event.companyId, taskId: event.payload.entityId })
+      .catch(err => logger.error({ err }, "post-commit chat completion delivery failed")));
+  });
+  server.on("close", unsubscribeChatCompletions);
   const runEnvironmentLeaseCleanupSweep = (backoffMs: number) =>
     environmentLeaseCleanupHeartbeat
       .sweepPendingCleanupLeases({ backoffMs })
@@ -1316,18 +1329,9 @@ async function startServerWithDatabaseTeardown(
       }));
   };
 
-  await chatCompletionDeliveries.sweepPending().catch((err) => logger.error({ err }, "startup chat completion delivery recovery failed"));
-  await connectionDeliveries.sweepPending();
   await app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed"));
   await app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed"));
-  await app.locals.toolActionDeliveries.sweepPending().catch((err: unknown) => logger.error({ err }, "startup tool review delivery sweep failed"));
-  await questionResponseDeliveries.sweepPending().then((result) => {
-    if (result.scanned > 0) {
-      logger.info(result, "startup question-response delivery sweep completed");
-    }
-  }).catch((err) => {
-    logger.error({ err }, "startup question-response delivery sweep failed");
-  });
+  await Promise.all(deliveryWorkers.map(worker => worker.ready));
   scheduleGitHubConnectionEventPoll();
   scheduleGitHubConnectionContinuitySweep();
 
@@ -1781,23 +1785,10 @@ async function startServerWithDatabaseTeardown(
             logger.error({ err }, "periodic secret proposal expiry sweep failed");
           }));
 
-        trackHeartbeatSchedulerWork(chatCompletionDeliveries.sweepPending().catch((err) => logger.error({ err }, "chat completion delivery failed")));
-
         trackHeartbeatSchedulerWork(accountingHeartbeat.reconcileCostAccounting().catch((err) => logger.error({ err }, "Cost accounting recovery failed")));
 
-        trackHeartbeatSchedulerWork(connectionDeliveries.sweepPending().catch((err) => logger.error({ err }, "connection continuation delivery failed")));
         trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "tool review recovery failed")));
         trackHeartbeatSchedulerWork(app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "gateway token cleanup failed")));
-        trackHeartbeatSchedulerWork(app.locals.toolActionDeliveries.sweepPending().catch((err: unknown) => logger.error({ err }, "tool review delivery sweep failed")));
-        trackHeartbeatSchedulerWork(questionResponseDeliveries.sweepPending()
-          .then((result) => {
-            if (result.scanned > 0) {
-              logger.info(result, "periodic question-response delivery sweep completed");
-            }
-          })
-          .catch((err) => {
-            logger.error({ err }, "periodic question-response delivery sweep failed");
-          }));
 
         if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
         if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) {
@@ -1974,6 +1965,7 @@ async function startServerWithDatabaseTeardown(
     await systemdNotify(["--stopping", `--status=Stopping after ${signal}`]);
     heartbeatSchedulerStopped = true;
     unsubscribeChatCompletions();
+    await deliveryWork.stop();
     clearInterval(executionControlInterval);
     if (heartbeatSchedulerInterval) {
       clearInterval(heartbeatSchedulerInterval);

server/src/routes/issues.ts

@@ -3384,14 +3384,6 @@ export function issueRoutes(
       ChatChannelService,
       "prepareFailedChatRunRetry" | "processFailedChatRunRetry"
     >;
-    feedbackExportService?: {
-      flushPendingFeedbackTraces(input?: {
-        companyId?: string;
-        traceId?: string;
-        limit?: number;
-        now?: Date;
-      }): Promise<unknown>;
-    };
     searchService?: CompanySearchService;
     searchRateLimiter?: CompanySearchRateLimiter;
     pluginWorkerManager?: PluginWorkerManager;
@@ -3647,7 +3639,6 @@ export function issueRoutes(
   const treeControlSvc = issueTreeControlFactory?.(db) ?? {
     getActivePauseHoldGate: async () => null,
   };
-  const feedbackExportService = opts?.feedbackExportService;
   const environmentsSvc = environmentService(db);
 
   async function queueTaskWatchdogEvaluation(
@@ -19005,21 +18996,8 @@ export function issueRoutes(
         );
       }
 
-      if (result.sharingEnabled && result.traceId && feedbackExportService) {
-        try {
-          await feedbackExportService.flushPendingFeedbackTraces({
-            companyId: issue.companyId,
-            traceId: result.traceId,
-            limit: 1,
-          });
-        } catch (err) {
-          logger.warn(
-            { err, issueId: issue.id, traceId: result.traceId },
-            "failed to flush shared feedback trace immediately",
-          );
-        }
-      }
-
+      // saveIssueVote has committed the export and notified its worker.
+      // Return the saved vote without waiting for an upload or backlog.
       res.status(201).json(result.vote);
     },
   );

server/src/services/chat-completion-delivery.ts

@@ -1,3 +1,4 @@
+import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { and, asc, eq, inArray, lte, sql } from "drizzle-orm";
 import { agents, agentWakeupRequests, chatCompletionDeliveries as deliveries, chatTaskHandoffs as handoffs,
   heartbeatRuns, issueComments, issueDocuments, issues, type Db } from "@paperclipai/db";
@@ -31,13 +32,16 @@ export async function recordChatHandoff(tx: Connection, task: Issue, actorRunId:
 }
 
 /** Must run on the same transaction as status projection (including native arbitration). */
-export async function recordChatCompletion(tx: Connection, before: Issue, after: Issue) {
-  if (before.status === after.status) return;
+export async function recordChatCompletion(tx: Connection, before: Issue, after: Issue): Promise<boolean> {
+  if (before.status === after.status) return false;
   await tx.update(deliveries).set({ status: "superseded" }).where(and(eq(deliveries.taskId, after.id),
     eq(deliveries.companyId, after.companyId), inArray(deliveries.status, [...pending])));
-  if (after.status !== "done") return;
+  if (after.status !== "done") return false;
   const [handoff] = await tx.select().from(handoffs).where(and(eq(handoffs.taskId, after.id), eq(handoffs.companyId, after.companyId)));
-  if (handoff) await tx.insert(deliveries).values({ companyId: after.companyId, taskId: after.id, statusVersion: after.statusVersion }).onConflictDoNothing();
+  if (!handoff) return false;
+  await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatCompletion);
+  await tx.insert(deliveries).values({ companyId: after.companyId, taskId: after.id, statusVersion: after.statusVersion }).onConflictDoNothing();
+  return true;
 }
 
 async function loadAudience(tx: Connection, deliveryId: string) {
@@ -243,5 +247,9 @@ export function chatCompletionDeliveryService(db: Db, heartbeat: { wakeup(agentI
       .orderBy(asc(deliveries.nextAttemptAt)).limit(100);
     for (const row of due) await deliver(row.id);
   }
-  return { deliver, sweepPending };
+  async function hasPending() {
+    return (await db.select({ id: deliveries.id }).from(deliveries)
+      .where(inArray(deliveries.status, [...pending])).limit(1)).length > 0;
+  }
+  return { deliver, sweepPending, hasPending };
 }

server/src/services/connection-intent-delivery.ts

@@ -167,7 +167,11 @@ export function connectionIntentDeliveryService(db: Db, heartbeat: Pick<Heartbea
       await db.update(connectionIntentDeliveries).set({ deliveredAt: new Date() }).where(eq(connectionIntentDeliveries.interactionId, interactionId));
     }
   }
-  return { deliver, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => {
+  async function hasPending() {
+    return (await db.select({ id: connectionIntentDeliveries.interactionId }).from(connectionIntentDeliveries)
+      .where(isNull(connectionIntentDeliveries.deliveredAt)).limit(1)).length > 0;
+  }
+  return { deliver, hasPending, tryDeliver: async (id: string) => { try { await deliver(id); } catch { /* Persisted delivery remains due after its lease. */ } }, sweepPending: async () => {
     const rows = await db.select().from(connectionIntentDeliveries).where(and(isNull(connectionIntentDeliveries.deliveredAt), lte(connectionIntentDeliveries.nextAttemptAt, new Date())))
       .orderBy(asc(connectionIntentDeliveries.nextAttemptAt)).limit(50);
     let failed = 0;

server/src/services/delivery-work-coordinator.ts

@@ -0,0 +1,100 @@
+import { subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "@paperclipai/db";
+import { createWorkScheduler } from "./work-scheduler.js";
+import { type DeliveryQueue } from "./delivery-work-notifications.js";
+import { beginIdleTrackedWork } from "./task-admission.js";
+
+export function createDeliveryWorkCoordinator(input: {
+  owner: object;
+  canRun: () => boolean;
+  // Idle drain may reconcile admitted writes without dispatching new work.
+  // Warm standby can disable both operations.
+  canReconcile?: () => boolean;
+  onError: (error: unknown, queue: DeliveryQueue) => void;
+}) {
+  const scheduler = createWorkScheduler();
+  const workers = new Map<DeliveryQueue, { wake: () => void; stop: () => Promise<void> }>();
+  let stopped = false;
+
+  return {
+    // Only these registered tasks are represented; not yet the whole instance.
+    nextWakeAt: scheduler.nextWakeAt,
+    register(queue: DeliveryQueue, task: {
+      retryMs: number;
+      run: (signal: AbortSignal) => Promise<unknown>;
+      hasPending: () => Promise<boolean>;
+    }) {
+      if (stopped) throw new Error("Delivery coordinator is stopped");
+      if (workers.has(queue)) throw new Error(`Delivery worker already registered: ${queue}`);
+      let dirty = false;
+      let scheduledAt: number | null = null;
+      let running: Promise<void> | null = null;
+      let workerStopped = false;
+      let controller: AbortController | null = null;
+      let finishIntent: (() => void) | undefined;
+      function schedule(delay: number) {
+        if (workerStopped) return;
+        const at = Date.now() + delay;
+        if (scheduledAt !== null && scheduledAt <= at) return;
+        scheduledAt = at;
+        scheduler.schedule(queue, at, start);
+      }
+      function start() {
+        if (workerStopped || running) return;
+        scheduler.cancel(queue);
+        scheduledAt = null;
+        if (!input.canRun() && !(input.canReconcile?.() ?? false)) { schedule(task.retryMs); return; }
+        dirty = false;
+        const finish = beginIdleTrackedWork();
+        const attempt = new AbortController();
+        controller = attempt;
+        running = Promise.resolve().then(() => reconcileDatabaseWork(input.owner, queue))
+          .then(() => workerStopped || !input.canRun() ? undefined : task.run(attempt.signal)).then(() => workerStopped ? false : task.hasPending()).then(pending => {
+          if (pending || databaseWorkPending(input.owner, queue)) schedule(task.retryMs);
+          else { finishIntent?.(); finishIntent = undefined; }
+        }).catch(error => {
+          schedule(task.retryMs);
+          // Logging must never turn a recoverable sweep failure into an
+          // unhandled rejection or prevent other queues from running.
+          try { if (!workerStopped) input.onError(error, queue); } catch { /* Recovery remains scheduled. */ }
+        }).finally(() => {
+          running = null;
+          controller = null;
+          finish();
+          if (dirty) schedule(0);
+        });
+      }
+      function wake() {
+        if (workerStopped) return;
+        dirty = true;
+        if (!running) schedule(0);
+      }
+      const unsubscribe = subscribeDatabaseWork(input.owner, queue, event => {
+        if (workerStopped) return;
+        if (databaseWorkPending(input.owner, queue)) {
+          finishIntent ??= beginIdleTrackedWork();
+          // Give the transaction time to settle. Its settlement notification
+          // replaces this deadline with an immediate wake on the normal path.
+          if (!running) schedule(task.retryMs);
+        }
+        if (event === "settled") wake();
+      });
+      const worker = { wake, async stop() {
+        workerStopped = true;
+        controller?.abort(new Error("Delivery worker stopped"));
+        unsubscribe();
+        scheduler.cancel(queue);
+        await running;
+        finishIntent?.(); finishIntent = undefined;
+      } };
+      workers.set(queue, worker);
+      const ready = Promise.resolve().then(() => { start(); return running; });
+      return { ready, wake };
+    },
+    async stop() {
+      stopped = true;
+      scheduler.stop();
+      await Promise.all(Array.from(workers.values(), worker => worker.stop()));
+      workers.clear();
+    },
+  };
+}

server/src/services/delivery-work-notifications.ts

@@ -0,0 +1,22 @@
+import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
+
+/** Topics for existing durable queues; register intent before their writes. */
+export const DELIVERY_QUEUES = {
+  feedback: "feedback-exports",
+  chatCompletion: "chat-completions",
+  connection: "connection-continuations",
+  question: "question-responses",
+  toolAction: "tool-action-receipts",
+} as const;
+export type DeliveryQueue = typeof DELIVERY_QUEUES[keyof typeof DELIVERY_QUEUES];
+
+export async function notifyDeliveryWork(transaction: object, queue: DeliveryQueue): Promise<void> {
+  await signalDatabaseWork(transaction, queue);
+}
+// Public delivery notification observers only see settled work. The coordinator
+// uses the lower-level lifecycle subscription to also fence in-flight writes.
+export function subscribeDeliveryWork(owner: object, queue: DeliveryQueue, wake: () => void): () => void {
+  return subscribeDatabaseWork(owner, queue, event => {
+    if (event === "settled") wake();
+  });
+}

server/src/services/feedback-share-client.ts

@@ -2,6 +2,8 @@ import { gzipSync } from "node:zlib";
 import type { FeedbackTraceBundle } from "@paperclipai/shared";
 import type { Config } from "../config.js";
 
+export const FEEDBACK_UPLOAD_TIMEOUT_MS = 30_000;
+
 const DEFAULT_FEEDBACK_EXPORT_BACKEND_URL = "https://telemetry.paperclip.ing";
 
 function buildFeedbackShareObjectKey(bundle: FeedbackTraceBundle, exportedAt: Date) {
@@ -12,7 +14,7 @@ function buildFeedbackShareObjectKey(bundle: FeedbackTraceBundle, exportedAt: Da
 }
 
 export interface FeedbackTraceShareClient {
-  uploadTraceBundle(bundle: FeedbackTraceBundle): Promise<{ objectKey: string }>;
+  uploadTraceBundle(bundle: FeedbackTraceBundle, signal?: AbortSignal): Promise<{ objectKey: string }>;
 }
 
 export function createFeedbackTraceShareClientFromConfig(
@@ -23,7 +25,10 @@ export function createFeedbackTraceShareClientFromConfig(
   const endpoint = new URL("/feedback-traces", baseUrl).toString();
 
   return {
-    async uploadTraceBundle(bundle) {
+    async uploadTraceBundle(bundle, shutdownSignal) {
+      const deadline = AbortSignal.timeout(FEEDBACK_UPLOAD_TIMEOUT_MS);
+      const signal = shutdownSignal ? AbortSignal.any([shutdownSignal, deadline]) : deadline;
+      signal.throwIfAborted();
       const exportedAt = new Date();
       const objectKey = buildFeedbackShareObjectKey(bundle, exportedAt);
       const requestBody = JSON.stringify({
@@ -33,6 +38,7 @@ export function createFeedbackTraceShareClientFromConfig(
       });
       const response = await fetch(endpoint, {
         method: "POST",
+        signal,
         headers: {
           "content-type": "application/json",
           ...(token ? { authorization: `Bearer ${token}` } : {}),
@@ -49,6 +55,8 @@ export function createFeedbackTraceShareClientFromConfig(
       }
 
       const payload = await response.json().catch(() => null) as { objectKey?: unknown } | null;
+      // A timed-out response body is not a successful upload acknowledgement.
+      signal.throwIfAborted();
       return {
         objectKey: typeof payload?.objectKey === "string" && payload.objectKey.trim().length > 0
           ? payload.objectKey

server/src/services/feedback.ts

@@ -1,3 +1,4 @@
+import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { readFile, readdir } from "node:fs/promises";
 import path from "node:path";
 import { and, asc, desc, eq, getTableColumns, gte, isNull, lte, ne, or } from "drizzle-orm";
@@ -111,7 +112,7 @@ const feedbackExportColumns = getTableColumns(feedbackExports);
 const instructionsSvc = agentInstructionsService();
 
 type FeedbackTraceShareClient = {
-  uploadTraceBundle(bundle: FeedbackTraceBundle): Promise<{ objectKey: string }>;
+  uploadTraceBundle(bundle: FeedbackTraceBundle, signal?: AbortSignal): Promise<{ objectKey: string }>;
 };
 
 type FeedbackServiceOptions = {
@@ -1715,6 +1716,15 @@ async function buildFeedbackTraceBundleFromRow(
 }
 
 export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
+  // Serialize explicit flushes and the worker's batches so callers cannot
+  // upload the same trace concurrently. HTTP vote handlers only save and notify.
+  let exportFlush: Promise<unknown> = Promise.resolve();
+  function serializeExportFlush<T>(operation: () => Promise<T>): Promise<T> {
+    const result = exportFlush.then(operation, operation);
+    exportFlush = result.catch(() => undefined);
+    return result;
+  }
+
   return {
     listIssueVotesForUser: async (issueId: string, authorUserId: string) =>
       db
@@ -1786,13 +1796,23 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
       return row ? buildFeedbackTraceBundleFromRow(db, row) : null;
     },
 
+    hasPendingFeedbackTraces: async () => {
+      const pending = options.shareClient
+        ? or(eq(feedbackExports.status, "pending"), eq(feedbackExports.status, "failed"))
+        : eq(feedbackExports.status, "pending");
+      return (await db.select({ id: feedbackExports.id }).from(feedbackExports)
+        .where(pending).limit(1)).length > 0;
+    },
+
     flushPendingFeedbackTraces: async (input?: {
       companyId?: string;
       traceId?: string;
       limit?: number;
       now?: Date;
-    }) => {
+      signal?: AbortSignal;
+    }) => serializeExportFlush(async () => {
       const shareClient = options.shareClient;
+      input?.signal?.throwIfAborted();
       if (!shareClient) {
         const filters = [eq(feedbackExports.status, "pending")];
         if (input?.companyId) {
@@ -1861,12 +1881,13 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
       let failed = 0;
 
       for (const row of rows) {
+        if (input?.signal?.aborted) break;
         const attemptAt = input?.now ?? new Date();
         attempted += 1;
 
         try {
           const bundle = await buildFeedbackTraceBundleFromRow(db, row);
-          await shareClient.uploadTraceBundle(bundle);
+          await shareClient.uploadTraceBundle(bundle, input?.signal);
 
           await db
             .update(feedbackExports)
@@ -1900,7 +1921,7 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
         sent,
         failed,
       };
-    },
+    }),
 
     saveIssueVote: async (input: {
       issueId: string;
@@ -2071,6 +2092,7 @@ export function feedbackService(db: Db, options: FeedbackServiceOptions = {}) {
           })
           .where(eq(feedbackVotes.id, savedVote.id));
 
+        if (sharedWithLabs) await notifyDeliveryWork(tx, DELIVERY_QUEUES.feedback);
         const [savedTrace] = await tx
           .insert(feedbackExports)
           .values({

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

@@ -0,0 +1,310 @@
+import { randomUUID } from "node:crypto";
+import type { Db } from "@paperclipai/db";
+import { agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, projects } from "@paperclipai/db";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import { runningProcesses } from "../../adapters/index.js";
+import { adapterExecutionControls } from "../adapter-execution-control.js";
+import { terminateLocalService } from "../local-service-supervisor.js";
+import { createHeartbeatRunState } from "./run-state.js";
+import { createHeartbeatRunPreparation } from "./run-preparation.js";
+import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "./run-lifecycle.js";
+import { createHeartbeatRunControl, type HeartbeatRunControlDependencies } from "./run-control.js";
+
+vi.mock("../live-events.js", () => ({ publishLiveEvent: vi.fn() }));
+vi.mock("../local-service-supervisor.js", async (importOriginal) => ({
+  ...await importOriginal<typeof import("../local-service-supervisor.js")>(),
+  terminateLocalService: vi.fn(async () => undefined),
+}));
+
+type Run = typeof heartbeatRuns.$inferSelect;
+function callbacks(db: Db) {
+  const state = createHeartbeatRunState(db);
+  const lifecycleDependencies = {
+    ...state,
+    ...createHeartbeatRunPreparation(db),
+    issuesSvc: {
+      addComment: vi.fn<HeartbeatLifecycleDependencies["issuesSvc"]["addComment"]>(),
+      listReviewAttention: vi.fn<HeartbeatLifecycleDependencies["issuesSvc"]["listReviewAttention"]>(),
+    },
+    recovery: { escalateStrandedAssignedIssue: vi.fn<HeartbeatLifecycleDependencies["recovery"]["escalateStrandedAssignedIssue"]>() },
+    companySkills: { completeTestRunForIssue: vi.fn<HeartbeatLifecycleDependencies["companySkills"]["completeTestRunForIssue"]>() },
+    budgets: { getInvocationBlock: vi.fn(async () => null) },
+    treeControlSvc: { getActivePauseHoldGate: vi.fn(async () => null) },
+    enqueueWakeup: vi.fn<HeartbeatLifecycleDependencies["enqueueWakeup"]>(),
+    getCurrentUserRedactionOptions: vi.fn(async () => ({})),
+    getAgentInvokability: vi.fn<HeartbeatLifecycleDependencies["getAgentInvokability"]>(async () => ({ invokable: true })),
+    emitTerminalAgentTaskRun: vi.fn<HeartbeatLifecycleDependencies["emitTerminalAgentTaskRun"]>(),
+    budgetHooks: {},
+  } satisfies HeartbeatLifecycleDependencies;
+  const lifecycle = createHeartbeatLifecycle(db, lifecycleDependencies);
+  return {
+    ...state,
+    ...lifecycle,
+    getRun: vi.fn(state.getRun),
+    getAgent: vi.fn(state.getAgent),
+    setRunStatusFromLive: vi.fn(lifecycle.setRunStatusFromLive),
+    setRunStatus: vi.fn(lifecycle.setRunStatus),
+    setWakeupStatus: vi.fn(lifecycle.setWakeupStatus),
+    appendRunEvent: vi.fn(lifecycle.appendRunEvent),
+    finalizeAgentStatus: vi.fn<HeartbeatRunControlDependencies["finalizeAgentStatus"]>(),
+    options: { closeWarmNativeSessionsForRun: vi.fn(async () => ({ closed: 1, busy: 0, failed: 0 })) },
+    envOrchestrator: { releaseForRun: vi.fn<HeartbeatRunControlDependencies["envOrchestrator"]["releaseForRun"]>(async () => ({ released: [], errors: [] })) },
+    enqueueWakeup: lifecycleDependencies.enqueueWakeup,
+    startNextQueuedRunForAgent: vi.fn<HeartbeatRunControlDependencies["startNextQueuedRunForAgent"]>(),
+    sweepPendingCleanupLeases: vi.fn<HeartbeatRunControlDependencies["sweepPendingCleanupLeases"]>(),
+    timerClaimWasFirstHeartbeat: vi.fn<HeartbeatRunControlDependencies["timerClaimWasFirstHeartbeat"]>(() => undefined),
+    wakeQueue: { releaseIssueExecution: vi.fn<HeartbeatRunControlDependencies["wakeQueue"]["releaseIssueExecution"]>(async () => ({ outcome: { kind: "released" }, postCommitEffects: [] })) },
+    applyWakeQueuePostCommitEffects: vi.fn(async () => undefined),
+    getSchedulingSuppression: vi.fn(async () => ({ suppressed: false })),
+    activeRunExecutions: new Set<string>(),
+    processRunCancellationSettlements: new Map<string, { settled: Promise<void>; failed: boolean; error?: unknown }>(),
+    failedProcessRunCancellations: new Map<string, { settled: Promise<void>; failed: boolean; error?: unknown }>(),
+  } satisfies HeartbeatRunControlDependencies;
+}
+function guardedDatabase() {
+  const access = vi.fn(() => { throw new Error("Unexpected run-control database access"); });
+  return { db: new Proxy({}, { get: access }) as Db, access };
+}
+
+describe("heartbeat run-control module boundary", () => {
+  afterEach(() => adapterExecutionControls.clear());
+
+  it("constructs without database access, cancellation, or cleanup", () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    createHeartbeatRunControl(database.db, deps);
+    expect(database.access).not.toHaveBeenCalled();
+    for (const callback of Object.values(deps)) if (vi.isMockFunction(callback)) expect(callback).not.toHaveBeenCalled();
+    expect(deps.envOrchestrator.releaseForRun).not.toHaveBeenCalled();
+    expect(deps.options.closeWarmNativeSessionsForRun).not.toHaveBeenCalled();
+  });
+
+  it("keeps an already completed run out of cancellation effects", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    const run = { id: "finished", status: "succeeded", runtimeMode: "legacy" } as Run;
+    deps.getRun.mockResolvedValue(run);
+    expect(await createHeartbeatRunControl(database.db, deps).cancelRunInternal(run.id)).toBe(run);
+    expect(database.access).not.toHaveBeenCalled();
+    expect(deps.getAgent).not.toHaveBeenCalled();
+    expect(deps.appendRunEvent).not.toHaveBeenCalled();
+  });
+
+  it("keeps a suppressed saved-input scan out of the database and wake admission", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    deps.getSchedulingSuppression.mockResolvedValue({ suppressed: true });
+    await createHeartbeatRunControl(database.db, deps).resumeExecutionWaitComments();
+    expect(database.access).not.toHaveBeenCalled();
+    expect(deps.enqueueWakeup).not.toHaveBeenCalled();
+  });
+
+  it("ignores an empty cancellation batch", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    expect(await createHeartbeatRunControl(database.db, deps).cancelInvocationsForAgentsInternal(["", ""], "pause")).toEqual({ agentIds: [], runsCancelled: 0, wakeupsCancelled: 0 });
+    expect(database.access).not.toHaveBeenCalled();
+    expect(deps.getAgent).not.toHaveBeenCalled();
+  });
+
+  it("waits for an owned adapter before admitting saved comments", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    const run = { id: "owned", status: "cancelled", runtimeMode: "legacy" } as Run;
+    adapterExecutionControls.set(run.id, {} as never);
+    await createHeartbeatRunControl(database.db, deps).resumeRemoteStopComments(run);
+    expect(database.access).not.toHaveBeenCalled();
+    expect(deps.enqueueWakeup).not.toHaveBeenCalled();
+  });
+
+  it("retains leases while native execution ownership is unverified", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    deps.getRun.mockResolvedValue({ runtimeMode: "native", status: "running", nativePhase: "terminal_failure", errorCode: "native_execution_ownership_unverified" } as Run);
+    await createHeartbeatRunControl(database.db, deps).releaseEnvironmentLeasesForRun({ runId: "held", companyId: "company", agentId: "agent", status: "failed", providerResourceDisposition: "destroy" });
+    expect(database.access).not.toHaveBeenCalled();
+    expect(deps.options.closeWarmNativeSessionsForRun).not.toHaveBeenCalled();
+    expect(deps.envOrchestrator.releaseForRun).not.toHaveBeenCalled();
+  });
+
+  it("uses the persisted terminal outcome when choosing warm resource retention", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    deps.getRun.mockResolvedValue({ status: "failed", runtimeMode: "legacy" } as Run);
+    await createHeartbeatRunControl(database.db, deps).releaseEnvironmentLeasesForRun({ runId: "failed", companyId: "company", agentId: "agent", status: "succeeded", providerResourceDisposition: "keep_running" });
+    expect(deps.envOrchestrator.releaseForRun).toHaveBeenCalledWith(expect.objectContaining({ heartbeatRunId: "failed", status: "failed", providerResourceDisposition: "stop_and_retain" }));
+    expect(database.access).not.toHaveBeenCalled();
+  });
+
+  it("defers destruction when a warm session remains busy", async () => {
+    const database = guardedDatabase();
+    const deps = callbacks(database.db);
+    deps.getRun.mockResolvedValue({ status: "failed", runtimeMode: "legacy" } as Run);
+    deps.options.closeWarmNativeSessionsForRun.mockResolvedValue({ closed: 0, busy: 1, failed: 0 });
+    await createHeartbeatRunControl(database.db, deps).releaseEnvironmentLeasesForRun({ runId: "busy", companyId: "company", agentId: "agent", status: "failed", providerResourceDisposition: "destroy" });
+    expect(deps.options.closeWarmNativeSessionsForRun).toHaveBeenCalledWith({ runId: "busy", reason: "terminal heartbeat run destroyed its environment lease" });
+    expect(deps.envOrchestrator.releaseForRun).not.toHaveBeenCalled();
+    expect(database.access).not.toHaveBeenCalled();
+  });
+});
+
+const support = await getEmbeddedPostgresTestSupport();
+if (!support.supported) console.warn(`Skipping run-control database tests: ${support.reason}`);
+describe.skipIf(!support.supported)("heartbeat run-control database wiring", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let db: Db;
+  beforeAll(async () => {
+    database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-run-control-module");
+    db = createDb(database.connectionString);
+  }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
+  afterAll(async () => {
+    await db?.$client.end();
+    await database?.cleanup();
+  });
+  beforeEach(async () => {
+    vi.mocked(terminateLocalService).mockReset().mockResolvedValue(undefined);
+    await db.execute(sql`truncate table companies restart identity cascade`);
+  });
+  afterEach(() => runningProcesses.clear());
+
+  async function fixture() {
+    const [company] = await db.insert(companies).values({ name: "Run control", issuePrefix: "RC" }).returning();
+    const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Controlled agent", adapterType: "codex_local" }).returning();
+    return { company, agent };
+  }
+  async function liveRun(companyId: string, agentId: string) {
+    const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running", runtimeMode: "legacy", runtimeModeResolvedAt: new Date() }).returning();
+    runningProcesses.set(run.id, { child: { pid: 424242 } as never, graceSec: 4, processGroupId: 424242 });
+    return run;
+  }
+
+  it("reports a missing run without entering cancellation", async () => {
+    const deps = callbacks(db);
+    await expect(createHeartbeatRunControl(db, deps).cancelRunInternal(randomUUID())).rejects.toThrow("Heartbeat run not found");
+    expect(deps.getAgent).not.toHaveBeenCalled();
+    expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
+    expect(deps.appendRunEvent).not.toHaveBeenCalled();
+  });
+
+  it("joins concurrent Stop requests across factories through the shared process barrier", async () => {
+    const { company, agent } = await fixture();
+    const run = await liveRun(company.id, agent.id);
+    const deps = callbacks(db);
+    const cancellationLookup = vi.spyOn(deps.processRunCancellationSettlements, "get");
+    let releaseTermination!: () => void;
+    const terminating = new Promise<void>(resolve => { releaseTermination = resolve; });
+    vi.mocked(terminateLocalService).mockReturnValue(terminating);
+    const first = createHeartbeatRunControl(db, deps).cancelRunInternal(run.id, "Stop", { terminationGraceMs: 1 });
+    await vi.waitFor(() => expect(deps.processRunCancellationSettlements.has(run.id)).toBe(true));
+    const second = createHeartbeatRunControl(db, deps).cancelRunInternal(run.id, "Stop again");
+    await vi.waitFor(() => expect(cancellationLookup).toHaveBeenCalledTimes(2));
+    expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
+    expect(terminateLocalService).toHaveBeenCalledTimes(1);
+    releaseTermination();
+    const results = await Promise.all([first, second]);
+    expect(results.map(result => result?.status)).toEqual(["cancelled", "cancelled"]);
+    expect(deps.setRunStatusFromLive).toHaveBeenCalledTimes(1);
+    expect(deps.appendRunEvent).toHaveBeenCalledTimes(1);
+    expect(deps.finalizeAgentStatus).toHaveBeenCalledTimes(1);
+    expect(deps.startNextQueuedRunForAgent).toHaveBeenCalledTimes(1);
+    expect(deps.processRunCancellationSettlements.size).toBe(0);
+    expect(runningProcesses.has(run.id)).toBe(false);
+    expect(terminateLocalService).toHaveBeenCalledWith({ pid: 424242, processGroupId: 424242 }, { forceAfterMs: 100, signal: "SIGINT" });
+  });
+
+  it("settles every joined Stop caller with the same termination failure", async () => {
+    const { company, agent } = await fixture();
+    const run = await liveRun(company.id, agent.id);
+    const deps = callbacks(db);
+    const cancellationLookup = vi.spyOn(deps.processRunCancellationSettlements, "get");
+    deps.activeRunExecutions.add(run.id);
+    let rejectTermination!: (error: Error) => void;
+    vi.mocked(terminateLocalService).mockReturnValue(new Promise<void>((_resolve, reject) => { rejectTermination = reject; }));
+    const first = createHeartbeatRunControl(db, deps).cancelRunInternal(run.id);
+    await vi.waitFor(() => expect(deps.processRunCancellationSettlements.has(run.id)).toBe(true));
+    const second = createHeartbeatRunControl(db, deps).cancelRunInternal(run.id);
+    const results = Promise.allSettled([first, second]);
+    await vi.waitFor(() => expect(cancellationLookup).toHaveBeenCalledTimes(2));
+    const failure = new Error("provider did not stop");
+    rejectTermination(failure);
+    expect(await results).toEqual([{ status: "rejected", reason: failure }, { status: "rejected", reason: failure }]);
+    expect(terminateLocalService).toHaveBeenCalledTimes(1);
+    expect(deps.processRunCancellationSettlements.size).toBe(0);
+    expect(deps.failedProcessRunCancellations.get(run.id)?.error).toBe(failure);
+    expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
+    expect((await deps.getRun(run.id))?.status).toBe("running");
+  });
+
+  it("retains failed Stop evidence and the owned process until a later request succeeds", async () => {
+    const { company, agent } = await fixture();
+    const run = await liveRun(company.id, agent.id);
+    const deps = callbacks(db);
+    deps.activeRunExecutions.add(run.id);
+    const failure = new Error("termination failed");
+    vi.mocked(terminateLocalService).mockRejectedValueOnce(failure);
+    await expect(createHeartbeatRunControl(db, deps).cancelRunInternal(run.id)).rejects.toBe(failure);
+    expect(deps.failedProcessRunCancellations.get(run.id)).toMatchObject({ failed: true, error: failure });
+    expect(deps.processRunCancellationSettlements.size).toBe(0);
+    expect(runningProcesses.has(run.id)).toBe(true);
+    expect((await deps.getRun(run.id))?.status).toBe("running");
+    expect(deps.finalizeAgentStatus).not.toHaveBeenCalled();
+    expect(await createHeartbeatRunControl(db, deps).cancelRunInternal(run.id)).toMatchObject({ status: "cancelled" });
+    expect(terminateLocalService).toHaveBeenCalledTimes(2);
+    // The executor still owns removal of this evidence after it observes Stop.
+    expect(deps.failedProcessRunCancellations.get(run.id)?.error).toBe(failure);
+  });
+
+  it("deduplicates agent batches and cancels only unclaimed pending wakes", async () => {
+    const { company, agent } = await fixture();
+    const [otherAgent] = await db.insert(agents).values({ companyId: company.id, name: "Other agent" }).returning();
+    const [finished] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, status: "succeeded" }).returning();
+    const wakes = await db.insert(agentWakeupRequests).values([
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "queued" },
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "deferred_issue_execution" },
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "claimed" },
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "queued", runId: finished.id },
+      { companyId: company.id, agentId: otherAgent.id, source: "assignment", status: "queued" },
+    ]).returning();
+    const deps = callbacks(db);
+    expect(await createHeartbeatRunControl(db, deps).cancelInvocationsForAgentsInternal([agent.id, "", agent.id], "pause")).toEqual({ agentIds: [agent.id], runsCancelled: 0, wakeupsCancelled: 2 });
+    const persisted = await db.select().from(agentWakeupRequests);
+    expect(wakes.map(wake => persisted.find(row => row.id === wake.id)?.status)).toEqual(["cancelled", "cancelled", "claimed", "queued", "queued"]);
+    expect(deps.getAgent).toHaveBeenCalledTimes(1);
+  });
+
+  it("rechecks stale budget enforcement before cancelling runs or queued wakes", async () => {
+    const { company, agent } = await fixture();
+    const [run] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, status: "queued", runtimeMode: "legacy" }).returning();
+    const [wake] = await db.insert(agentWakeupRequests).values({ companyId: company.id, agentId: agent.id, source: "assignment", status: "queued" }).returning();
+    const deps = callbacks(db);
+    await createHeartbeatRunControl(db, deps).cancelBudgetScopeWork({ companyId: company.id, scopeType: "agent", scopeId: agent.id, enforcement: { policyId: randomUUID(), version: 1 } });
+    expect((await deps.getRun(run.id))?.status).toBe("queued");
+    expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0].status).toBe("queued");
+    expect(deps.setRunStatusFromLive).not.toHaveBeenCalled();
+    expect(deps.finalizeAgentStatus).not.toHaveBeenCalled();
+    expect(terminateLocalService).not.toHaveBeenCalled();
+  });
+
+  it("limits project budget cancellation by company, effective project, and admission cutoff", async () => {
+    const { company, agent } = await fixture();
+    const [otherCompany] = await db.insert(companies).values({ name: "Other company", issuePrefix: "OC" }).returning();
+    const [otherAgent] = await db.insert(agents).values({ companyId: otherCompany.id, name: "Other company agent" }).returning();
+    const projectRows = await db.insert(projects).values([
+      { companyId: company.id, name: "Target" }, { companyId: company.id, name: "Override" },
+    ]).returning();
+    const [issue] = await db.insert(issues).values({ companyId: company.id, projectId: projectRows[0].id, title: "Bound task", issueNumber: 1, identifier: "RC-1" }).returning();
+    const old = new Date("2026-01-01T00:00:00Z");
+    const cutoff = new Date("2026-01-02T00:00:00Z");
+    const wakes = await db.insert(agentWakeupRequests).values([
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "queued", createdAt: old, payload: { issueId: issue.id } },
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "queued", createdAt: old, payload: { issueId: issue.id, projectId: projectRows[1].id } },
+      { companyId: company.id, agentId: agent.id, source: "assignment", status: "queued", createdAt: cutoff, payload: { projectId: projectRows[0].id } },
+      { companyId: otherCompany.id, agentId: otherAgent.id, source: "assignment", status: "queued", createdAt: old, payload: { projectId: projectRows[0].id } },
+    ]).returning();
+    await createHeartbeatRunControl(db, callbacks(db)).cancelBudgetScopeWork({ companyId: company.id, scopeType: "project", scopeId: projectRows[0].id, createdBefore: cutoff });
+    const persisted = await db.select().from(agentWakeupRequests);
+    expect(wakes.map(wake => persisted.find(row => row.id === wake.id)?.status)).toEqual(["cancelled", "queued", "queued", "queued"]);
+  });
+});

server/src/services/issue-thread-interactions.ts

@@ -1,4 +1,5 @@
 import { normalizeEscapedLineBreaks } from "@paperclipai/shared/validators/text";
+import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { activeIssueInteractionCondition, historicalQuestionCondition } from "./issue-question-context.js";
 import {
   currentContinuationOrigins,
@@ -2829,6 +2830,7 @@ export function issueThreadInteractionService(
         .returning();
         if (!row) throw interactionAlreadyResolvedError();
         if (status === "accepted" || status === "rejected") {
+          await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection);
           await tx.insert(connectionIntentDeliveries).values({ interactionId, companyId: issue.companyId }).onConflictDoNothing();
         }
         return row;
@@ -5026,6 +5028,7 @@ export function issueThreadInteractionService(
         // This answer updates conversation history only. It must not resume
         // the completed source run or enqueue new work for the closed task.
         if (!historicalAnswer) {
+          await notifyDeliveryWork(tx, DELIVERY_QUEUES.question);
           await tx
             .insert(issueQuestionResponseDeliveries)
             .values(questionResponseDeliveryValues(answered));

server/src/services/question-response-delivery.ts

@@ -1340,5 +1340,9 @@ export function questionResponseDeliveryService(
     return counts;
   }
 
-  return { deliver, sweepPending };
+  async function hasPending() {
+    return (await db.select({ id: issueQuestionResponseDeliveries.id }).from(issueQuestionResponseDeliveries)
+      .where(inArray(issueQuestionResponseDeliveries.status, ["pending", "delivering"])).limit(1)).length > 0;
+  }
+  return { deliver, sweepPending, hasPending };
 }

server/src/services/tool-action-delivery.ts

@@ -345,6 +345,12 @@ export function toolActionDeliveryService(
   }
   return {
     deliver,
+    async hasPending() {
+      // Include receipts awaiting review/execution: their later terminal write
+      // must remain recoverable even if its optional fast-path callback fails.
+      return (await db.select({ id: toolActionDeliveries.actionRequestId }).from(toolActionDeliveries)
+        .where(isNull(toolActionDeliveries.deliveredAt)).limit(1)).length > 0;
+    },
     async deliverForRun(input: { companyId: string; runId: string }) {
       // A review can settle before its source run yields. Retry that run's
       // receipts as soon as execution cleanup finishes; the periodic sweep

server/src/services/tool-action-review.ts

@@ -1,3 +1,4 @@
+import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { and, eq } from "drizzle-orm";
 import {
   issues,
@@ -206,6 +207,7 @@ export async function commitToolActionReview(
               : { version: 1, outcome: "rejected", reason: input.reason },
         })
         .where(eq(issueThreadInteractions.id, interaction.id));
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolAction);
       await tx
         .insert(toolActionDeliveries)
         .values({

server/src/services/tool-gateway.ts

@@ -1,3 +1,4 @@
+import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js";
 import { composeConnectionInstructions } from "./connection-instructions.js";
 import { isInsufficientConnectionScope, INSUFFICIENT_CONNECTION_SCOPE_MESSAGE } from "./connection-permission-errors.js";
 import { boundedMcpToolName } from "./mcp-tool-names.js";
@@ -2615,15 +2616,18 @@ export function createToolGatewayService(
       );
     }
 
-    await db
-      .insert(toolActionDeliveries)
-      .values({
-        companyId: input.session.companyId,
-        actionRequestId: actionRequest.id,
-        issueId: input.session.issueId,
-        interactionId: interaction.id,
-      })
-      .onConflictDoNothing();
+    await db.transaction(async tx => {
+      await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolAction);
+      await tx
+        .insert(toolActionDeliveries)
+        .values({
+          companyId: input.session.companyId,
+          actionRequestId: actionRequest.id,
+          issueId: input.session.issueId!,
+          interactionId: interaction.id,
+        })
+        .onConflictDoNothing();
+    });
 
     await writeToolCallEvent({
       invocationId: input.invocation.id,

server/src/services/work-scheduler.ts

@@ -0,0 +1,45 @@
+/** In-process deadlines. Owners restore durable tasks at startup. */
+export function createWorkScheduler() {
+  const tasks = new Map<string, { at: number; run: () => void }>();
+  let timer: ReturnType<typeof setTimeout> | null = null;
+  let stopped = false;
+  const nextWakeAt = () => tasks.size ? Math.min(...Array.from(tasks.values(), task => task.at)) : null;
+
+  function arm() {
+    if (timer) clearTimeout(timer);
+    timer = null;
+    const next = nextWakeAt();
+    if (stopped || next === null) return;
+    timer = setTimeout(() => {
+      timer = null;
+      const now = Date.now();
+      const due = Array.from(tasks.entries()).filter(([, task]) => task.at <= now);
+      for (const [key, task] of due) {
+        // Another due callback can cancel or replace this task.
+        if (tasks.get(key) !== task) continue;
+        tasks.delete(key);
+        try { task.run(); }
+        catch (error) { process.emitWarning(`Scheduled work failed: ${String(error)}`); }
+      }
+      arm();
+    }, Math.min(2_147_483_647, Math.max(0, next - Date.now())));
+    timer.unref?.();
+  }
+
+  return {
+    nextWakeAt,
+    schedule(key: string, at: number, run: () => void) {
+      if (stopped) return;
+      if (!Number.isFinite(at)) throw new Error("Work deadline must be finite");
+      tasks.set(key, { at, run });
+      arm();
+    },
+    cancel(key: string) { tasks.delete(key); arm(); },
+    stop() {
+      stopped = true;
+      tasks.clear();
+      if (timer) clearTimeout(timer);
+      timer = null;
+    },
+  };
+}