fix(server): wake decision continuations from committed changes (#17478)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A completed decision can owe a wake to the agent that proposed it. > - The decision row already stores the pending continuation. > - The main maintenance tick scans this debt even when no continuation is ready. > - Committed changes can activate recovery and leave settled history quiet. > - This pull request gives continuation delivery its own worker and shares foreground delivery ownership. > - Pending work stays recoverable without an unconditional continuation scan. ## Linked Issues or Issue Description Targets master after #17475 merged. This is slice 09 of the recurring server work migration. Each PR owns one reason for background work. The remaining inventory is in `doc/plans/2026-10-24-recurring-server-sleep-stack.md`. **What existing behavior does this improve?** Decision continuation delivery and recovery in the server. **Current behavior** Foreground paths send a wake and clear a pending flag. The combined decision sweep scans pending continuations on every maintenance tick. It can overlap foreground delivery. **Proposed behavior** Terminal decision writes notify a dedicated worker through the existing transaction boundary. Foreground and recovery calls share one in-flight delivery per decision and DB owner. Recovery re-reads readiness, pages past failures, and retries only unresolved debt or errors. **Reason and benefit** Open decisions, running effects, and delivered history no longer cause an independent continuation scan. No queue table is added. Decision expiry, target invalidation, interrupted effects, and retention remain separate duties. ## What Changed - Extract continuation readiness, delivery, and paged recovery into the decision owner. Preserve the wake payload and at-least-once crash behavior. - Signal the ten decision update sites before SQL in their existing transaction scope. Keep effects and wake calls outside those transactions. - Explicitly create the continuation obligation on all three archive-proposal failure settlements, including recovered running rows without inherited pending metadata. - Serialize concurrent decision effect execution through final settlement. Coalesce foreground and worker wakes. Re-read the durable flag before sending. Fence effect completion to running state so competing completion cannot restore acknowledged debt. - Expose an opaque DB work-scope key. Child pools and nested transactions share that identity; independent roots remain separate. Existing signal consumers keep their behavior. - Bound each recovery pass to 100 candidates, retain a fair cursor across failures, and stop between deliveries on shutdown. Forward the existing AbortSignal through the shared deadline adapter. - Register enabled-heartbeat continuation recovery before decision housekeeping. Remove only its independent scan from the main tick. Preserve the public combined sweep and disabled-heartbeat behavior. - Add generated migration `0388_ambitious_micromax.sql`, an idempotent concurrent partial index for pending continuations, and keep five generated snapshots. - Add real producer, transaction, failure, concurrency, paging, quiet, and shutdown tests. Document ownership and remaining limits in `doc/idle-decision-continuations.md`. ## Verification Current candidate: `18f682863cadb91281a86b4da2553a4579ee054a`, normally integrated with landed archive squash `5785090ff1`. All 31 independent incoming master files are preserved exactly, including controller-authority recovery. All 97 decision/startup/deadline cases plus server types/build pass after the required Runner vendor rebuild. The preceding full five-suite selection passed 293 cases; the continuation owner behavior is unchanged by master reconciliation. - **97 cases pass at this master integration** across decision, startup and scoped-deadline suites, plus server types/build. The preceding parent candidate passed the full 293-case decision/gateway/archive/startup/deadline selection; those results remain evidence for that revision. All selected cases executed with separate PostgreSQL fixtures where needed. - Tenant-driven global scan amplification reproduced six sweeps for six sequential real cancellations. Existing thirty-second notification coalescing now permits one recovery sweep per window while retaining pending work. The first green fixture wrongly expected the intent hold to clear before coalesced reconciliation; the hold remains and quiet is checked after settlement. Foreground delivery stays immediate. - The archive contention test uses a signaled dedicated single-connection read and two FIFO SQL round trips instead of a fixed delay. Removing effect coalescing makes it fail with two contender reads; restoring production code passes. No production test hook was added. - **Three new real PostgreSQL regressions failed before repair**: recovered archive proposals with invalid signatures, mismatched manifests or archive execution failures and no inherited pending flag. Each now persists failure, wakes once, acknowledges continuation delivery and remains quiet on replay. Running-state fencing is preserved. - The unmerged continuation index is regenerated as `0388_ambitious_micromax.sql` after parent migrations 386/387. All parent journal entries are unchanged, five generated snapshots remain, numbering/safety and snapshot/schema-drift checks pass. Fresh database migration apply and replay pass; the concurrent index is valid. - Existing lifecycle coverage includes foreground/worker coalescing, child pools, transaction commit/rollback, failed acknowledgement, restart, drain/resume, shutdown and fair 100-row recovery. Earlier bounded-pass and concurrent archive-result regressions and their repairs remain recorded in [the owning behavior audit](doc/idle-decision-continuations.md). - Failure accounting: the first combined command used an incorrect retention fixture variable and skipped 15 cases. The corrected separate external-PostgreSQL retention run passed all 15. The historical nine embedded-only work-signal cases were not executed by this selection. Extra upstream embedded-only suites attempted on the parent could not initialize locally; they remain CI verification, as recorded in the parent PR. Earlier reused-fixture, assertion and compile failures remain in the behavior audit and stack ledger; they are not erased by later passing checks. Commands: `pnpm --filter @paperclipai/server typecheck`, `pnpm --filter @paperclipai/server build`, server Vitest for `decisions-service`, `decision-retention-service`, `server-startup-feedback-export` and `scoped-deadline-work`; `pnpm --filter @paperclipai/db check:migrations`, migration apply/replay, and the generated snapshot-drift test. Local targeted evidence does not establish a full local repository run, deployed sleep or fleet savings. Fresh exact-head CI and both automated reviews remain merge gates. ## Risks - Notification-driven recovery remains global but is coalesced to thirty seconds. This bounds request-driven scans; it does not introduce company-scoped selection. Foreground delivery remains immediate. - Delivery remains at least once. A crash or acknowledgement failure after accepting a wake can send it again. Local coalescing is not a cross-process claim. - Supported writers share the application DB signal owner. Independent SQL writers or multiple active server processes need a separate wake contract. Wake-producing service methods do not gain a new guarantee when called inside an uncommitted caller transaction. - The concurrent index needs top-level migration execution; it is registered with the existing migration owner and can be replayed safely. - No Cloud sleep blocker or external alarm coverage changes. Worker quiet does not prove deployed app or database suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-28 20:21:50 +01:00 · 2026-10-27 17:44:51 -05:00
1 parent 5785090ff1
commit a63d19816f
@@ -0,0 +1,193 @@
+# Decision continuation delivery while idle
+
+A terminal decision can owe an agent wake. Its existing row stores that debt:
+`continuationPolicy = wake_origin_agent`, `metadata.continuationPending = true`,
+and either expired/cancelled status or a decided status with succeeded/partial/
+failed execution. Open decisions and running effects are not ready to deliver.
+This migration adds no queue table and changes no wake payload.
+
+## Ownership and producers
+
+`decisions.ts` owns decision creation, authorization, signatures, effects, and
+state changes. Its ten update sites now use a narrow transaction-aware write
+helper before the SQL statement. These cover the running claim; ordinary effect
+completion; archive-proposal signature, manifest, and execution failures; archive
+success; TTL rejection on use; cancel; direct dismissal; and background expiry.
+The helper acquires intent in the existing outer transaction or its savepoint.
+It does not include effect execution, provider calls, activity publication, or
+agent wake callbacks in a new database transaction.
+
+Creation and bundle creation insert open decisions, so they do not create ready
+continuation debt. Acknowledgement only clears debt and belongs to the delivery
+owner; it does not signal itself. The production decision services are created
+by server startup and routes with the application's DB handle. Independent SQL
+writers and additional active processes require a separate notification and
+ownership contract. This change does not make wake-producing service methods
+safe to call inside an uncommitted caller transaction; only their SQL mutation
+boundary has the existing outer-commit guarantee.
+
+`decision-continuations.ts` owns the readiness predicate, delivery, and recovery.
+Each worker pass attempts at most 100 candidates. A saved in-memory cursor
+advances past failed candidates between passes and wraps after the last page.
+Remaining durable work retains another pass. This avoids starving later rows behind a failing first page.
+A partial ID index contains ready pending continuations only. The predicate uses
+fixed SQL literals so a prepared generic plan can still use that index.
+
+The shared DB package exposes an opaque work-scope identity. Raw DB object
+identity is insufficient because dedicated and liveness child pools already
+share their owner's signals. The key is shared by those children and nested
+transactions; independent roots have different keys. No connection credentials,
+private transaction state, or decision policy are exposed. Existing signal
+consumers and transactions that do not register work keep their behavior.
+
+## Foreground, recovery, and failure
+
+Foreground delivery remains immediate. All service instances sharing the DB
+owner coalesce a simultaneous wake for the same decision into one promise.
+Delivery re-reads durable readiness after claiming that local flight, so a late
+foreground call cannot resend a continuation already acknowledged by recovery.
+The promise is removed on success or failure. Effect execution and its final
+state write also share one local flight per decision. This prevents concurrent
+archive callers from turning a successful archive into a stale-manifest failure. This is process-local coalescing,
+not a new durable claim or an exactly-once guarantee.
+
+Effect completion requires the row to remain running. A competing completion
+cannot restore pending metadata after another caller has completed and delivered
+it. Effects retain their own existing durable claims and authorization checks.
+The delivery worker never executes effects.
+
+A wake failure leaves pending metadata intact. A failed acknowledgement also
+leaves debt, so a later attempt can send the wake again. A crash after accepting
+the wake but before acknowledgement retains the existing at-least-once contract.
+An uncertain producer transaction remains held until the shared work-signal
+protocol establishes settlement; one empty scan cannot clear that uncertainty.
+
+## Lifecycle and scope
+
+Startup reconstructs pending delivery. Committed changes wake the worker through
+the existing coordinator. Failures or remaining ready debt retain a 30-second
+retry. Open/running decisions, no decisions, and delivered history have no
+recurring continuation scan or deadline. Changes during a pass retain another
+request. Drain/standby defer dispatch without consuming that request; shutdown
+waits for the current admitted delivery and stops before starting another.
+The shared deadline adapter forwards its existing AbortSignal to task execution;
+existing callbacks that do not consume the signal keep their behavior. A future process-local retry is not a Cloud alarm.
+
+Enabled-heartbeat startup registers this worker before the combined decision
+sweep. The main tick skips that sweep's independent continuation scan. Its public
+API retains combined behavior by default. Expiry or effect recovery can still
+deliver their newly produced continuations immediately. Disabled-heartbeat
+startup keeps its existing behavior and does not dispatch this new worker.
+
+Decision expiry, invalidated-target scanning, interrupted effect recovery, and
+retention remain separate migrations. Existing Cloud safety blockers are not
+relaxed. This establishes worker quiet under the named conditions, not deployed
+host/database suspension or measured fleet savings.
+
+## Verification
+
+The selected run passes 93 tests: 50 real-PostgreSQL decision cases, 39 startup
+cases, and four database work-signal unit cases. Fifteen new continuation cases
+cover real decide/cancel/dismiss/expiry producers, running-to-terminal completion,
+foreground contention, dedicated child pools, stale caller re-read, restart,
+failed acknowledgement, outer rollback/commit, first write during drain, failed
+retry through drain, changes during delivery, shutdown, and paging past a failed
+candidate. Quiet checks observe no SQL calls across an hour of controlled time.
+Nine pre-existing embedded-PostgreSQL signal tests are skipped locally because
+the embedded runtime is unavailable; CI must run that supported environment.
+Workspace typecheck and build pass. The final literal-predicate adjustment also
+passes all 49 decision tests before the additional archive-race regression.
+
+The original continuation index applied and replayed without error. After parent integration it is generated migration `0388_ambitious_micromax.sql`; the fresh apply and replay also pass. A rolled-back synthetic
+40,000-row history with one pending continuation uses the partial index under
+`force_generic_plan`, reads two shared buffers, and executes in 0.015 ms on the
+local PostgreSQL server. This is query evidence, not a production latency claim.
+
+The first test selection had 90 passes and one failure: the injected PostgreSQL
+acknowledgement error is wrapped by Drizzle, but the assertion expected its
+message at the top level. The assertion now checks the original cause. No
+production retry, hold, or timeout was weakened. A recursive unit-fixture type
+initially prevented migration generation; an explicit fixture type fixed it.
+Full CI and fresh review remain required before PR readiness.
+
+
+Review found two real regressions. Before repair, a controlled concurrent archive
+reported failed after archiving successfully, and shutdown sent a second wake
+after stop was requested. Both regressions failed before the fix. Shared local
+effect ownership fixes archive completion; bounded passes and abort propagation
+fix backlog shutdown. The 93-case selection passes after repair. The paging test
+also proves a failed first candidate cannot hide the 101st candidate across
+bounded passes. A result-type alias initially made the known decision ID optional;
+the alias now matches the existing outcome contract.
+
+The existing tool-review and archive-delivery consumers also pass 124 real-DB
+tests after AbortSignal propagation. Together with the decision/startup/signal
+selection, 217 selected tests pass; the same nine older embedded-only cases
+remain locally skipped. Workspace typecheck passes after the outcome type fix.
+
+
+## Explicit archive-failure delivery
+
+The three archive-proposal failure paths now explicitly set
+`continuationPending: true` for `wake_origin_agent`, rather than assuming the
+pre-read running row already carries that metadata. The existing running-state
+fence remains, so stale settlement cannot re-arm an acknowledged result.
+Invalid signatures, manifest mismatches and archive execution failures each
+failed a real-PostgreSQL recovery regression before repair. Each now saves the
+failed result, wakes once, acknowledges delivery and remains quiet on replay.
+
+Normal parent integration preserves the projection interaction-link index and
+archive pending-order index. The unmerged continuation index was regenerated
+as 0388 from the parent's generated metadata; every parent journal entry is
+unchanged, five snapshots remain, the SQL uses the existing top-level concurrent
+executor, and snapshot/schema drift passes. No generated snapshot was hand-edited.
+
+The final affected selection passes 111 cases: 53 decisions, 15 retention/archive,
+42 startup and one scoped-deadline case, with all four suites exercised. The
+first combined command used the wrong retention DB variable and skipped those
+15 cases; the corrected external-PostgreSQL retention run passes all 15. Server
+typecheck and build pass. The historical nine embedded-only work-signal skips
+above are not reclassified as executed by this selection. Fresh CI and reviews
+remain required.
+
+
+## Bounded notification-driven recovery
+
+Security review identified that sequential decision writes could each start an
+instance-wide continuation scan. A real-producer regression reproduced six
+sweeps for six sequential cancellations. The worker now uses the existing
+`notificationCoalesceMs: 30_000` option. The first wake after a quiet window still
+reconciles promptly, and subsequent notifications in that window share its end
+deadline. Foreground continuation delivery is unchanged and immediate. Recovery
+can wait up to thirty seconds, matching the prior maintenance cadence. This
+bounds global scan frequency; it does not make queries company-scoped or add a
+cross-process notification contract.
+
+The regression verifies one scan during repeated real create/cancel writes,
+retention and eventual delivery of a later pending row, and no recurring SQL
+after settlement. Its first green attempt wrongly expected zero active work
+while the coalesced request was pending. The existing intent hold correctly
+remains until reconciliation; the test now checks that hold's actual terminal
+boundary through the existing quiet assertion. No production hold was removed.
+All 293 integrated cases across decisions, archive, gateway/projection, startup
+and scoped-deadline suites pass without skips; server types/build pass. Fresh
+security review, Greptile, Ponytail and exact-head CI remain merge gates.
+
+
+The archive contention regression no longer waits a fixed 100 ms. A dedicated
+single-connection contender signals its decision read; two FIFO SQL round trips
+then establish that it reached shared execution while the first archive remains
+held. The test requires one contender decision read and one archive execution.
+Temporarily removing the effect-flight guard made this test fail (two reads);
+restoring the production guard passes. The mutation was restored before commit.
+This verifies actual cross-pool contention without a timing delay or a new
+production-only test hook.
+
+
+Archive #17475 is merged. Normal reconciliation with its exact squash
+`5785090ff1` preserves all 31 independent incoming master files, including
+Runner controller-authority recovery. All 97 decision/startup/deadline cases and
+server types/build pass after that integration. The Runner vendor rebuilt from
+its changed inputs. The preceding full five-suite selection passed 293 cases;
+the continuation owner's production behavior is unchanged by reconciliation.
+Fresh exact-head reviews and CI remain required on the master-target candidate.
@@ -15,10 +15,12 @@ Do not mark an item done from a timer removal or a green check alone.
 | 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | [#17443](https://github.com/paperclipai/paperclip/pull/17443), merged after CI and passing reviews |
 | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending |
 | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending |
-| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; notification delivery implemented in slice 08, PR pending |
-| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending |
+| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged after exact-head 5/5 reviews and passing current CI |
+| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478), expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479), and effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) await fresh current-head gates; target invalidation review repair remains in progress |
+
 | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending |
-| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), merged after current-head CI and both 5/5 reviews; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), merged after current-head CI and both 5/5 reviews; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), interaction-link index verified and awaiting fresh CI/review |
+| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Slices 02–07 merged: [secret proposals](https://github.com/paperclipai/paperclip/pull/17446), [credentials](https://github.com/paperclipai/paperclip/pull/17450), [review expiry](https://github.com/paperclipai/paperclip/pull/17453), [approved recovery](https://github.com/paperclipai/paperclip/pull/17454), [interrupted claims](https://github.com/paperclipai/paperclip/pull/17458), [projection](https://github.com/paperclipai/paperclip/pull/17468) |
+
 | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending |
 | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending |
 | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending |
@@ -359,3 +361,131 @@ counter regression and corrected disabled-mode expectation are recorded in the
 behavior document. Full current-head CI and both reviews remain required.
 Retention eligibility/scanning, decision expiry, effect recovery, and decision
 continuation delivery remain independent unfinished reasons in the full ledger.
+
+
+Slice 08 is [#17475](https://github.com/paperclipai/paperclip/pull/17475), head
+`692e3ee02c`, based on #17468. The PR is attached and Greptile review was
+requested. Fresh CI is active with no failures in the latest snapshot.
+The first seven PRs are mergeable and have all checks passing at their recorded
+heads. They remain open; no merge was requested for this stack.
+
+## Next owner audit: decision continuations
+
+The next branch is `codex/sleep-09-decision-continuations`, based on #17475.
+No ninth implementation or readiness is claimed. `decisions.ts:sweepExpired`
+currently mixes four distinct reasons: interrupted effect recovery, pending
+continuation delivery, TTL expiry, and invalidated strict targets. Preserve
+separate PRs for those duties. A deadline alone cannot replace target-change
+notifications; target deletion and completion need an audited producer contract.
+
+For continuation delivery, the existing decision row already stores the durable
+obligation: `continuationPolicy = wake_origin_agent`,
+`metadata.continuationPending = true`, and a terminal outcome. Decided rows
+are ready only after execution reaches succeeded, partial, or failed. Open
+decisions and still-running effects must not create a delivery retry.
+
+All current decision-row writers are in `decisions.ts`. Production service
+instances are created in server startup and decision routes with the server DB
+handle. Audit creation/bundles, decision claims, regular and archive-proposal
+effect settlement/failure, TTL expiry on use, background expiry, cancellation,
+dismissal, and continuation acknowledgement. Reuse a narrow pre-write helper
+inside existing transaction ownership; do not wrap provider/effect execution
+or agent wake calls in a database transaction. The archive-proposal error
+branches inherit the claim's pending metadata and must remain eligible after
+terminal failure.
+
+The current foreground paths call `deliverContinuation` after saving the result.
+It sends the wake then clears pending metadata. Simply adding an eager worker
+would make foreground/worker overlap common. Preserve prompt foreground delivery
+and the existing at-least-once crash contract, but coalesce simultaneous attempts
+for the same decision and re-read durable readiness before sending. Scope any
+in-memory coalescing to the actual database/instance owner; do not let unrelated
+instances or differing transaction ownership share an accidental global key.
+The DB signal owner also supports child pools, so a proposed coalescing mechanism
+must explicitly reconcile that boundary rather than assuming object identity
+is always sufficient. No new table is needed merely for this migration.
+
+After commit-aware producer coverage, add a paged continuation-only recovery
+entry point and register it with the existing coordinator. Remove only its
+independent pending scan from the periodic combined sweep. Keep normal
+continuations produced by expiry/effect completion, and preserve the public
+combined recovery entry point for current callers/tests. A failed wake or
+acknowledgement retains its durable pending state and a retry. Matching delivered
+history becomes quiet.
+
+Tests should exercise actual decide/cancel/dismiss/expiry producers, completion
+of effects after a running claim, concurrent foreground and worker delivery,
+wake/acknowledgement failures, restart, outer commit/rollback, first write and
+failure during drain, pagination past failing rows, and delivered-history quiet.
+The existing decisions suite has an embedded-only setup; use the repository's
+validated external test database helper to exercise it on the disposable local
+PostgreSQL server, without silently skipping the suite when embedded setup is
+unavailable.
+
+Slice 08 review update: Greptile is 5/5 on `692e3ee02c` with no actionable
+findings. The latest CI snapshot has 10 successful checks and 44 in progress,
+with no failures. This is a live CI wait, not a readiness or completion claim.
+
+
+## Slice 09 implementation: decision continuations
+
+The continuation owner and worker are implemented on `codex/sleep-09-decision-continuations`.
+See [decision continuations](../idle-decision-continuations.md) for producer,
+transaction, concurrency, failure, and test evidence. The existing row is durable
+truth; the only DB API addition is an opaque shared work-scope identity for local
+coalescing across child pools. No new queue table or external wake exemption is
+added. Terminal effect settlement is fenced to running state so a competing
+caller cannot re-arm an acknowledged continuation. A concurrent partial index
+bounds pending selection independently of retained history.
+
+The selected run passes 92 tests; nine older embedded-Postgres cases are locally
+skipped, with the new lifecycle tests running against real external PostgreSQL.
+Workspace typecheck/build pass. Migration apply/replay and a 40,000-row prepared
+query probe pass. Initial fixture type/error-shape failures are recorded in the
+behavior document. PR creation and full CI/review are next.
+
+Parent #17475 at `692e3ee02c` now has passing checks and Greptile 5/5; the extra
+Paperclip Review check was skipped. Fresh master is `fe85fd7ac4`, still at
+migration 0377. The stack remains open and unmerged. The full inventory above
+remains the completion scope.
+
+
+Slice 09 is [#17478](https://github.com/paperclipai/paperclip/pull/17478), initially
+`66bbef521e`. Review found a concurrent archive result race and unbounded backlog
+shutdown. Both failed real-boundary regressions before repair. The local repair
+shares effect execution through completion per DB owner/decision, bounds each
+continuation pass to 100 with a fair cursor, and forwards cancellation through
+the shared deadline adapter. The repaired selection passes 217 tests, with nine
+older embedded-only signal tests skipped locally. Typecheck passes. Fresh build,
+CI, and review remain required; the initial head has no other CI failure.
+
+The separate TTL branch `codex/sleep-10-decision-expiry` contains implementation
+commit `18c11e3110`. It passed 101 selected tests, workspace typecheck and build.
+Its fault fixture initially cleared its error collector before a previous
+admitted failure finished; the assertion now verifies all errors after settled
+recovery. No hold or retry was weakened. This branch must receive the parent
+review repairs and adapt its own paged expiry worker to cancellation before PR
+publication. No tenth PR or readiness is claimed.
+
+
+Slice 09 review follow-up: every failed archive-proposal settlement explicitly
+creates its origin-agent continuation obligation. Three recovered-row regressions
+failed before repair and pass with exactly one wake and persisted acknowledgement.
+The current parent-integrated selection passes 111 decision/archive/startup/deadline
+cases, server types/build, fresh migration apply/replay and generated snapshot
+drift. Own unmerged index is regenerated as `0388_ambitious_micromax.sql`; parent
+386/387 migrations and journal entries are preserved. The initial skipped retention
+command and corrected 15-case run are recorded in the behavior document.
+
+Slice 09 security follow-up: sequential real decision writes reproduced six
+global continuation scans. Existing thirty-second notification coalescing now
+bounds those scans while preserving immediate foreground wakes and durable
+pending recovery. All 293 integrated decision/archive/gateway/startup/deadline
+cases and server types/build pass. The incorrect expectation of zero active
+work during the coalescing window is recorded in the behavior document; the
+intent hold remains until reconciliation. Fresh security/CI/reviews required.
+
+Eight slices are merged after archive delivery #17475. Slice 09 now reconciles
+with its exact squash `5785090ff1`; all 31 independent master files are preserved.
+The 97 decision/startup/deadline cases and server types/build pass after the
+Runner vendor rebuild. Fresh master-target CI and reviews remain required.
@@ -506,6 +506,8 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
  "0385_open_madame_web.sql",
  "0386_steady_epoch.sql",
  "0387_boring_professor_monster.sql",
+  "0388_ambitious_micromax.sql",
+

  "0384_funny_adam_warlock.sql",
  "0383_tool_review_company_expiry.sql",
@@ -45,4 +45,4 @@ 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";
+export { databaseWorkScopeKey, signalDatabaseWork, subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "./work-signals.js";
@@ -0,0 +1 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "decisions_pending_continuation_idx" ON "decisions" USING btree ("id") WHERE "decisions"."continuation_policy" = 'wake_origin_agent' AND "decisions"."metadata" ->> 'continuationPending' = 'true' AND ("decisions"."status" IN ('expired', 'cancelled') OR ("decisions"."status" = 'decided' AND "decisions"."execution_status" IN ('succeeded', 'partial', 'failed')));
@@ -1,6 +1,6 @@
 {
-  "id": "410457fc-266b-4bd0-8bc0-2b14b0634d07",
-  "prevId": "ecec138e-5312-4ad6-b051-bd4ee9ed4e34",
+  "id": "4f21a636-735f-4dce-9ab8-0583f3bc7b79",
+  "prevId": "de27fdf8-430a-443a-9502-725f92c48892",
  "version": "7",
  "dialect": "postgresql",
  "tables": {
@@ -22110,6 +22110,34 @@
          "concurrently": false,
          "method": "btree",
          "with": {}
+        },
+        "decision_archive_notification_outbox_delivery_idx": {
+          "name": "decision_archive_notification_outbox_delivery_idx",
+          "columns": [
+            {
+              "expression": "last_attempt_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "first"
+            },
+            {
+              "expression": "created_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"decision_archive_notification_outbox\".\"status\" = 'pending'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
        }
      },
      "foreignKeys": {
@@ -24030,6 +24058,22 @@
          "method": "btree",
          "with": {}
        },
+        "decisions_pending_continuation_idx": {
+          "name": "decisions_pending_continuation_idx",
+          "columns": [
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"decisions\".\"continuation_policy\" = 'wake_origin_agent' AND \"decisions\".\"metadata\" ->> 'continuationPending' = 'true' AND (\"decisions\".\"status\" IN ('expired', 'cancelled') OR (\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" IN ('succeeded', 'partial', 'failed')))",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "decisions_bundle_idx": {
          "name": "decisions_bundle_idx",
          "columns": [
@@ -52856,6 +52900,44 @@
          "method": "btree",
          "with": {}
        },
+        "tool_action_requests_executing_deadline_idx": {
+          "name": "tool_action_requests_executing_deadline_idx",
+          "columns": [
+            {
+              "expression": "updated_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"tool_action_requests\".\"status\" = 'executing'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "tool_action_requests_approved_idx": {
+          "name": "tool_action_requests_approved_idx",
+          "columns": [
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"tool_action_requests\".\"status\" = 'approved'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "tool_action_requests_invocation_idx": {
          "name": "tool_action_requests_invocation_idx",
          "columns": [
@@ -52871,6 +52953,27 @@
          "method": "btree",
          "with": {}
        },
+        "tool_action_requests_company_interaction_idx": {
+          "name": "tool_action_requests_company_interaction_idx",
+          "columns": [
+            {
+              "expression": "company_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "interaction_id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
        "tool_action_requests_issue_idx": {
          "name": "tool_action_requests_issue_idx",
          "columns": [
@@ -2703,6 +2703,13 @@
      "when": 1793136115698,
      "tag": "0387_boring_professor_monster",
      "breakpoints": true
+    },
+    {
+      "idx": 388,
+      "version": "7",
+      "when": 1793137983484,
+      "tag": "0388_ambitious_micromax",
+      "breakpoints": true
    }
  ]
 }
@@ -69,6 +69,8 @@ export const decisions = pgTable(
      table.status,
      table.expiresAt,
    ),
+    pendingContinuationIdx: index("decisions_pending_continuation_idx").on(table.id)
+      .where(sql`${table.continuationPolicy} = 'wake_origin_agent' AND ${table.metadata} ->> 'continuationPending' = 'true' AND (${table.status} IN ('expired', 'cancelled') OR (${table.status} = 'decided' AND ${table.executionStatus} IN ('succeeded', 'partial', 'failed')))`),
    bundleIdx: index("decisions_bundle_idx").on(table.bundleId),
    originIssueIdx: index("decisions_origin_issue_idx").on(table.originIssueId),
    companyIdempotencyUq: uniqueIndex("decisions_company_idempotency_uq")
@@ -3,7 +3,7 @@ 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";
+import { databaseWorkScopeKey, databaseWorkPending, installDatabaseWorkSignals, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js";

 const support = await getEmbeddedPostgresTestSupport();
 (support.supported ? describe : describe.skip)("database work signals", () => {
@@ -187,3 +187,23 @@ describe("work signal query boundaries", () => {
    await expect(signalDatabaseWork(connection, "bad")).rejects.toThrow("inside a transaction");
  });
 });
+
+
+describe("database work scope identity", () => {
+  type TestConnection = { execute: () => Promise<unknown[]>; transaction: (callback: (tx: TestConnection) => unknown) => Promise<unknown> };
+  function connection(): TestConnection {
+    return { execute: vi.fn(async () => []), transaction: async callback => callback(connection()) };
+  }
+  it("shares owner identity through child pools and nested transactions, but isolates independent owners", async () => {
+    const first = installDatabaseWorkSignals(connection());
+    const second = installDatabaseWorkSignals(connection());
+    const child = installDatabaseWorkSignals(connection(), first);
+    const key = databaseWorkScopeKey(first);
+    expect(databaseWorkScopeKey(second)).not.toBe(key);
+    expect(databaseWorkScopeKey(child)).toBe(key);
+    await child.transaction(async tx => {
+      expect(databaseWorkScopeKey(tx)).toBe(key);
+      await tx.transaction(async nested => { expect(databaseWorkScopeKey(nested)).toBe(key); });
+    });
+  });
+});
@@ -5,7 +5,7 @@ import { sql } from "drizzle-orm";
 // an XID, so unrelated transactions keep their existing SQL behavior.
 type Listener = (event: "intent" | "settled", hints?: ReadonlySet<string>) => void;
 type Transaction = { topics: Set<string>; hints: Map<string, Set<string>>; xid?: Promise<string>; uncertain: boolean };
-type Scope = { listeners: Map<string, Set<Listener>>; transactions: Set<Transaction>; root: Connection };
+type Scope = { key: object; 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>();
@@ -19,7 +19,7 @@ function publish(scope: Scope, topic: string, event: "intent" | "settled" = "set

 /** 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 };
+  const scope = (owner && contexts.get(owner)?.scope) || { key: {}, listeners: new Map(), transactions: new Set(), root: db };
  install(db, { scope });
  return db;
 }
@@ -112,3 +112,7 @@ export function subscribeDatabaseWork(owner: object, topic: string, listener: Li
  }
  return () => { listeners.delete(listener); };
 }
+
+export function databaseWorkScopeKey(connection: object): object {
+  return contexts.get(connection)?.scope.key ?? connection;
+}
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
 import { mkdtempSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import path from "node:path";
-import { eq, inArray } from "drizzle-orm";
+import { eq, inArray, sql } from "drizzle-orm";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import {
  activityLog,
@@ -11,6 +11,7 @@ import {
  companies,
  companyMemberships,
  createDb,
+  withDedicatedDbConnection,
  decisionEffectExecutions,
  decisionRetention,
  decisions,
@@ -23,10 +24,19 @@ import {
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { attentionService } from "../services/attention.js";
 import { decisionService } from "../services/decisions.js";
+import * as retentionModule from "../services/decision-retention.js";
 import { hashAttentionArchiveManifest } from "../services/decision-retention.js";
 import type { AttentionArchiveManifestEntry, AttentionArchiveTargetSnapshot } from "@paperclipai/shared";

-const support = await getEmbeddedPostgresTestSupport();
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { decisionContinuationService } from "../services/decision-continuations.js";
+import { registerDecisionContinuationWork } from "../services/decision-continuation-work.js";
+import { writeDecisionChange } from "../services/decision-work-notifications.js";
+import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISIONS_TEST_DATABASE_URL");
+const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describePg = support.supported ? describe : describe.skip;

 describePg("decisionService", () => {
@@ -41,9 +51,11 @@ describePg("decisionService", () => {
  let decidedByUserId: string;
  let wakes: Array<Record<string, unknown>>;

+  let stopWorker: (() => Promise<void>) | undefined;
+
  beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decisions-");
-    db = createDb(tempDb.connectionString);
+    if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decisions-");
+    db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
  }, 20_000);

  beforeEach(async () => {
@@ -69,13 +81,14 @@ describePg("decisionService", () => {
  });

  afterEach(async () => {
+    await stopWorker?.(); stopWorker = undefined;
    delete process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE;
    delete process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS;
    delete process.env.PAPERCLIP_AGENT_JWT_SECRET;
    await db.delete(decisionEffectExecutions); await db.delete(decisionTargetIssues); await db.delete(decisions); await db.delete(decisionRetention); await db.delete(activityLog);
    await db.delete(issueComments); await db.delete(issueRelations); await db.delete(heartbeatRuns); await db.delete(issues); await db.delete(agents); await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies);
  });
-  afterAll(async () => tempDb?.cleanup());
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); });

  const agentActor = () => ({ type: "agent" as const, companyId, agentId, runId, source: "agent_jwt" as const,
    onBehalfOfUserId: originResponsibleUserId, onBehalfOfMemberships: [{ companyId, membershipRole: "member", status: "active" }] });
@@ -194,6 +207,87 @@ describePg("decisionService", () => {
    expect(states.find((row) => row.sourceId === unreviewedIssueId)?.archivedAt).toBeNull();
  });

+  it.each(["invalid_signature", "manifest_mismatch", "archive_failure"] as const)("delivers a recovered archive failure without inherited pending metadata: %s", async failure => {
+    const activityAt = new Date("2026-04-01T00:00:00.000Z");
+    const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetIssueId,
+      expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }];
+    const row = await service().create({ companyId, actor: agentActor(), agentId, runId, title: "Archive?", body: "Exact set",
+      continuationPolicy: "wake_origin_agent", options: [{ id: "archive", label: "Archive", effects: [] }],
+      metadata: { kind: "attention_archive_proposal", manifestHash: failure === "manifest_mismatch" ? "incorrect" : hashAttentionArchiveManifest(manifest) },
+      additionalTargetSnapshots: { [`attention:review:${targetIssueId}`]: { status: "attention", assigneeAgentId: null,
+        assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } });
+    await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "archive",
+      decidedByUserId, decidedAt: new Date(), updatedAt: new Date(Date.now() - 120_000),
+      ...(failure === "invalid_signature" ? { signedSpec: "invalid" } : {}),
+      metadata: { ...row.metadata, continuationPending: false } }).where(eq(decisions.id, row.id));
+    const original = retentionModule.decisionRetentionService;
+    const factory = failure === "archive_failure" ? vi.spyOn(retentionModule, "decisionRetentionService").mockImplementation((source, options) => ({
+      ...original(source, options), archiveReviewedManifest: async () => { throw new Error("archive_failure"); },
+    })) : undefined;
+    try {
+      expect(await service().sweepExpired()).toMatchObject({ resumed: 1 });
+      const saved = await service().get(row.id);
+      expect(saved).toMatchObject({ executionStatus: "failed", metadata: { archiveProposalError: failure, continuationPending: false } });
+      expect(saved!.metadata.continuationDeliveredAt).toEqual(expect.any(String));
+      expect(wakes).toHaveLength(1);
+      await service().sweepExpired();
+      expect(wakes).toHaveLength(1);
+    } finally { factory?.mockRestore(); }
+  });
+
+  it("serializes concurrent archive completion before reporting its outcome", async () => {
+    await db.update(issues).set({ status: "in_review", assigneeUserId: decidedByUserId }).where(eq(issues.id, targetIssueId));
+    const activityAt = new Date("2026-04-01T00:00:00.000Z");
+    await db.insert(decisionRetention).values({ companyId, sourceKind: "review", sourceId: targetIssueId, sourceActivityAt: activityAt });
+    const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetIssueId,
+      expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }];
+    const row = await service().create({ companyId, actor: agentActor(), agentId, runId, title: "Archive?", body: "Exact set",
+      continuationPolicy: "wake_origin_agent", options: [{ id: "archive", label: "Archive", effects: [] }],
+      metadata: { kind: "attention_archive_proposal", manifestHash: hashAttentionArchiveManifest(manifest) },
+      additionalTargetSnapshots: { [`attention:review:${targetIssueId}`]: { status: "attention", assigneeAgentId: null,
+        assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } });
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const original = retentionModule.decisionRetentionService;
+    let archived = false, calls = 0;
+    const factory = vi.spyOn(retentionModule, "decisionRetentionService").mockImplementation((source, options) => {
+      const actual = original(source, options);
+      return { ...actual, archiveReviewedManifest: async input => {
+        calls++;
+        const result = await actual.archiveReviewedManifest(input);
+        archived = true; await gate; return result;
+      } };
+    });
+    const decide = () => service().decide({ id: row.id, optionId: "archive", decidedByUserId, userActor: boardActor() });
+    const first = decide();
+    let second: ReturnType<typeof decide> | undefined;
+    try {
+      await vi.waitFor(() => expect(archived).toBe(true));
+      await withDedicatedDbConnection(db, async contender => {
+        let observed!: () => void;
+        const readStarted = new Promise<void>(resolve => { observed = resolve; });
+        let decisionReads = 0;
+        contender.$client.options.debug = (_id, query) => {
+          if (/^select/i.test(query) && query.includes('from "decisions"')) { decisionReads++; observed(); }
+        };
+        try {
+          second = decisionService(contender, { wakeOriginAgent: async input => { wakes.push(input); } })
+            .decide({ id: row.id, optionId: "archive", decidedByUserId, userActor: boardActor() });
+          await readStarted;
+          await contender.execute(sql`select 1`);
+          await contender.execute(sql`select 1`);
+          expect(decisionReads).toBe(1);
+          release();
+          const results = await Promise.all([first, second]);
+          expect(results.map(result => result.executionStatus)).toEqual(["succeeded", "succeeded"]);
+          expect(calls).toBe(1); expect(wakes).toHaveLength(1);
+          const [state] = await db.select().from(decisionRetention);
+          expect(state.archiveVersion).toBe(1); expect(state.archivedAt).not.toBeNull();
+        } finally { release(); await Promise.allSettled([first, ...(second ? [second] : [])]); }
+      });
+    } finally { release(); await Promise.allSettled([first, ...(second ? [second] : [])]); factory.mockRestore(); }
+  });
+
  it("skips strict stale targets and fails closed on intersection denial", async () => {
    const stale = await createCommentDecision("strict");
    await db.update(issues).set({ updatedAt: new Date(Date.now() + 1_000) }).where(eq(issues.id, targetIssueId));
@@ -656,4 +750,240 @@ describePg("decisionService", () => {
    expect(result).toMatchObject({ status: "decided", chosenOptionId: "dismissed" });
    expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]);
  });
+  describe("continuation worker", () => {
+    let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+    let draining: boolean;
+    const errors: unknown[] = [];
+    const wake = vi.fn(async (_input: Record<string, unknown>) => {});
+    const workerService = () => decisionService(db, { wakeOriginAgent: wake });
+
+    beforeEach(async () => {
+      await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+      draining = false; errors.length = 0; wake.mockReset();
+      stopWorker = async () => {
+        await coordinator?.stop(); coordinator = undefined;
+        vi.useRealTimers(); vi.restoreAllMocks();
+        await db.execute(sql`drop trigger if exists decision_ack_fail on decisions`);
+        await db.execute(sql`drop function if exists decision_ack_fail()`);
+        expect(errors).toEqual([]);
+      };
+      vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+    });
+    async function start(svc = workerService()) {
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
+        canReconcile: () => true, onError: error => errors.push(error) });
+      await registerDecisionContinuationWork(coordinator, svc, () => !draining).ready;
+    }
+    async function quiet() {
+      await vi.waitFor(async () => {
+        await vi.advanceTimersByTimeAsync(30_000);
+        expect(idleWorkSnapshot().active).toBe(0);
+        expect(coordinator!.nextWakeAt()).toBeNull();
+      }, { timeout: 5000 });
+      const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute");
+      await vi.advanceTimersByTimeAsync(60 * 60_000);
+      expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
+      reads.mockRestore(); writes.mockRestore(); executes.mockRestore();
+    }
+    async function pending(id: string) {
+      expect((await service().get(id))?.metadata.continuationPending).toBe(true);
+    }
+    async function delivered(id: string) {
+      expect((await service().get(id))?.metadata.continuationPending).toBe(false);
+    }
+    const markReady = (id: string, source = db) => writeDecisionChange(source, async tx => tx.update(decisions)
+      .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id)));
+
+    it("coalesces sequential tenant writes without dropping pending recovery", async () => {
+      const row = await createCommentDecision();
+      const svc = workerService(), sweep = vi.spyOn(svc, "sweepContinuations");
+      await start(svc); await quiet(); sweep.mockClear();
+      await svc.cancel(row.id, { actorType: "agent", actorId: agentId });
+      await vi.advanceTimersByTimeAsync(1000);
+      await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
+      expect(sweep).toHaveBeenCalledTimes(1);
+      for (let i = 0; i < 5; i++) {
+        const next = await createCommentDecision();
+        await svc.cancel(next.id, { actorType: "agent", actorId: agentId });
+        await vi.advanceTimersByTimeAsync(1000);
+      }
+      expect(sweep).toHaveBeenCalledTimes(1);
+      const recovery = await createCommentDecision();
+      await markReady(recovery.id);
+      await vi.advanceTimersByTimeAsync(1000);
+      await pending(recovery.id);
+      expect(sweep).toHaveBeenCalledTimes(1);
+      await quiet(); await delivered(recovery.id);
+      expect(sweep).toHaveBeenCalledTimes(2);
+      expect(wake.mock.calls.filter(([input]) => input.decisionId === recovery.id)).toHaveLength(1);
+    });
+
+    it.each(["decide", "cancel", "dismiss", "expire"] as const)("reactivates from real %s and leaves delivered history quiet", async action => {
+      const row = await createCommentDecision();
+      await start(); await quiet();
+      const svc = workerService();
+      if (action === "decide") await svc.decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() });
+      if (action === "cancel") await svc.cancel(row.id, { actorType: "agent", actorId: agentId });
+      if (action === "dismiss") await svc.dismiss(row.id, decidedByUserId, boardActor());
+      if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepExpired(new Date(), { includeContinuations: false }); }
+      await quiet(); await delivered(row.id);
+      expect(wake).toHaveBeenCalledTimes(1);
+      expect(wake.mock.calls[0][0]).toMatchObject({ decisionId: row.id, companyId, agentId, issueId: originIssueId });
+    });
+
+    it("stays quiet while open or running, then wakes when effect completion commits", async () => {
+      const row = await createCommentDecision();
+      await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes",
+        decidedByUserId, metadata: { continuationPending: true } }).where(eq(decisions.id, row.id));
+      await start(); await quiet(); expect(wake).not.toHaveBeenCalled();
+      await workerService().decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() });
+      await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1);
+      expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1);
+    });
+
+    it("coalesces simultaneous foreground replay and worker delivery without replaying effects", async () => {
+      const row = await createCommentDecision();
+      await start();
+      await Promise.all(Array.from({ length: 3 }, () => workerService().decide({
+        id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor(),
+      }).catch(error => { if (error.message !== "decision_already_resolved") throw error; })));
+      await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1);
+      expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1);
+    });
+
+    it("shares an in-flight wake with a dedicated child pool and rereads after completion", async () => {
+      const row = await createCommentDecision(); await markReady(row.id);
+      let release!: () => void;
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      wake.mockImplementationOnce(async () => gate);
+      const first = decisionContinuationService(db, wake).deliver(row.id);
+      const childWake = vi.fn(async () => {});
+      let child: Promise<void>;
+      try {
+        await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1));
+        child = withDedicatedDbConnection(db, async connection => {
+          const attempt = decisionContinuationService(connection, childWake).deliver(row.id);
+          expect(attempt).toBe(first);
+          await attempt;
+        });
+        await vi.advanceTimersByTimeAsync(100);
+      } finally { release(); }
+      await first; await child!;
+      await decisionContinuationService(db, childWake).deliver(row.id);
+      expect(childWake).not.toHaveBeenCalled(); await delivered(row.id);
+    });
+
+    it("reconstructs pending delivery at restart without replaying decision effects", async () => {
+      const row = await createCommentDecision();
+      wake.mockRejectedValueOnce(new Error("wake unavailable"));
+      await expect(workerService().decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() })).rejects.toThrow("wake unavailable");
+      await pending(row.id); await start(); await quiet(); await delivered(row.id);
+      expect(wake).toHaveBeenCalledTimes(2);
+      expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1);
+    });
+
+    it("retries a failed acknowledgement with the existing at-least-once wake contract", async () => {
+      const row = await createCommentDecision(); await start();
+      await db.execute(sql`create function decision_ack_fail() returns trigger language plpgsql as $$ begin
+        if NEW.metadata ->> 'continuationPending' = 'false' then raise exception 'ack unavailable'; end if;
+        return NEW; end $$`);
+      await db.execute(sql`create trigger decision_ack_fail before update on decisions for each row execute function decision_ack_fail()`);
+      await expect(workerService().cancel(row.id, { actorType: "agent", actorId: agentId })).rejects.toMatchObject({ cause: { message: "ack unavailable" } });
+      await pending(row.id);
+      await vi.advanceTimersByTimeAsync(60_000);
+      await db.execute(sql`drop trigger decision_ack_fail on decisions`);
+      await quiet(); await delivered(row.id);
+      expect(wake.mock.calls.length).toBeGreaterThan(1);
+      for (const error of errors) expect(error).toMatchObject({ cause: { message: "ack unavailable" } });
+      errors.length = 0;
+    });
+
+    it("waits for an outer transaction, ignores rollback, and recovers the committed write", async () => {
+      const row = await createCommentDecision(); await start(); await quiet();
+      await expect(db.transaction(async tx => {
+        await markReady(row.id, tx as unknown as typeof db);
+        await vi.advanceTimersByTimeAsync(60_000);
+        expect(wake).not.toHaveBeenCalled(); throw new Error("rollback");
+      })).rejects.toThrow("rollback");
+      await quiet(); expect((await service().get(row.id))?.status).toBe("open");
+      await db.transaction(async tx => {
+        await markReady(row.id, tx as unknown as typeof db);
+        await vi.advanceTimersByTimeAsync(60_000); expect(wake).not.toHaveBeenCalled();
+      });
+      await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1);
+    });
+
+    it("retains a first write during drain without scanning and delivers after resume", async () => {
+      const row = await createCommentDecision(); await start(); await quiet(); draining = true;
+      await markReady(row.id);
+      const reads = vi.spyOn(db, "select");
+      await vi.advanceTimersByTimeAsync(60_000);
+      expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); expect(wake).not.toHaveBeenCalled();
+      draining = false; await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1);
+    });
+
+    it("retains a failed wake through drain and resumes its retry", async () => {
+      const row = await createCommentDecision(); await start();
+      wake.mockImplementationOnce(async () => { draining = true; throw new Error("wake unavailable"); });
+      await markReady(row.id);
+      await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors).toHaveLength(1); });
+      await pending(row.id);
+      const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000);
+      expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+      draining = false; errors.length = 0; await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(2);
+    });
+
+    it("preserves changes arriving during delivery and stops after shutdown", async () => {
+      const firstRow = await createCommentDecision(), secondRow = await createCommentDecision();
+      let release!: () => void;
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      wake.mockImplementationOnce(async () => { await markReady(secondRow.id); await gate; });
+      await markReady(firstRow.id);
+      const starting = start();
+      try { await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); }
+      finally { release(); }
+      await starting; await quiet(); await delivered(firstRow.id); await delivered(secondRow.id);
+      expect(wake).toHaveBeenCalledTimes(2);
+      await coordinator!.stop(); coordinator = undefined;
+      await vi.advanceTimersByTimeAsync(60_000); expect(wake).toHaveBeenCalledTimes(2);
+    });
+
+    it("awaits an admitted wake before completing shutdown", async () => {
+      const row = await createCommentDecision(), other = await createCommentDecision();
+      await markReady(row.id); await markReady(other.id);
+      let release!: () => void;
+      const gate = new Promise<void>(resolve => { release = resolve; });
+      wake.mockImplementationOnce(async () => gate);
+      const starting = start();
+      let stopped = false;
+      let stopping: Promise<void>;
+      try {
+        await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1));
+        stopping = coordinator!.stop().then(() => { stopped = true; });
+        await vi.advanceTimersByTimeAsync(60_000);
+        expect(stopped).toBe(false);
+      } finally { release(); }
+      await starting; await stopping!; coordinator = undefined;
+      expect(wake).toHaveBeenCalledTimes(1);
+      const sentId = wake.mock.calls[0][0].decisionId as string;
+      await delivered(sentId); await pending(sentId === row.id ? other.id : row.id); expect(stopped).toBe(true);
+    });
+
+    it("pages past a failed first candidate and preserves its retry", async () => {
+      const seed = await createCommentDecision();
+      await db.delete(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, seed.id));
+      await db.delete(decisions).where(eq(decisions.id, seed.id));
+      const ids = Array.from({ length: 101 }, (_, i) => `10000000-0000-4000-8000-${String(i).padStart(12, "0")}`);
+      await db.insert(decisions).values(ids.map(id => ({ ...seed, id, status: "cancelled", metadata: { continuationPending: true } })));
+      wake.mockImplementation(async input => { if (input.decisionId === ids[0]) throw new Error("first unavailable"); });
+      const recovery = workerService();
+      await expect(recovery.sweepContinuations()).rejects.toThrow("first unavailable");
+      expect(wake).toHaveBeenCalledTimes(100); await pending(ids.at(-1)!);
+      expect(await recovery.sweepContinuations()).toEqual({ retry: true });
+      expect(wake).toHaveBeenCalledTimes(101); await delivered(ids.at(-1)!); await pending(ids[0]);
+      wake.mockImplementation(async () => {});
+      await start(); await quiet(); await delivered(ids[0]); expect(wake).toHaveBeenCalledTimes(102);
+    });
+  });
+
 });
@@ -6,7 +6,8 @@ import { readIdleSleepSafety } from "../services/idle-sleep-safety.js";
 import * as idleLocalWork from "../services/idle-local-work.js";
 import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js";
 import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
-import { decisionRetentionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js";
+import { decisionService, decisionRetentionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js";
+
 import { companies, runDatabaseBackup, type Db } from "@paperclipai/db";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs";
@@ -348,6 +349,7 @@ vi.mock("../services/index.js", () => ({
  })),
  decisionService: vi.fn(() => ({
    sweepExpired: vi.fn(async () => ({ expired: 0 })),
+    sweepContinuations: vi.fn(async () => ({ retry: false })),
  })),
  decisionRetentionService: vi.fn(() => ({
    autoArchive: vi.fn(async () => 0),
@@ -913,6 +915,10 @@ describe("startServer feedback export wiring", () => {
      expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1);
      expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1);
      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.customImageExpiry)).toBe(true);
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(true);
+      const decisions = vi.mocked(decisionService).mock.results.at(-1)!.value;
+      expect(decisions.sweepContinuations).toHaveBeenCalledTimes(1);
+      expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false });
    } finally {
      setIntervalSpy.mockRestore();
    }
@@ -949,6 +955,7 @@ describe("startServer feedback export wiring", () => {
      expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled();
      expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
      expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled();
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(false);
    } finally {
      setIntervalSpy.mockRestore();
    }
@@ -1,3 +1,5 @@
+import { registerDecisionContinuationWork } from "./services/decision-continuation-work.js";
+
 import { registerDecisionArchiveDeliveryWork } from "./services/decision-archive-delivery-work.js";

 import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js";
@@ -1835,7 +1837,9 @@ async function startServerWithDatabaseTeardown(
          logger.warn({ ...toolHealthSweep }, "startup tool connection health sweep found failing connections");
        }
      });
-      await step("decision_expiry", () => decisionExecutor.sweepExpired());
+      await step("decision_continuation_worker", () => registerDecisionContinuationWork(deliveryWork, decisionExecutor,
+        () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
+      await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false }));
      const loginSchedule = loginCleanupSchedule(db, config.heartbeatSchedulerIntervalMs);
      await step("login_cleanup_workers", () => registerLoginCleanupWork(deliveryWork, [
        { queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult), next: loginSchedule.adapter },
@@ -1856,7 +1860,7 @@ async function startServerWithDatabaseTeardown(
      trackHeartbeatSchedulerWork((async () => {
        if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
        scheduleRoutineStatusReconcile();
-        trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired().catch((err: unknown) => {
+        trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false }).catch((err: unknown) => {
          logger.error({ err }, "decision expiry sweep failed");
        }));
        trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => {
@@ -5,7 +5,7 @@ export function registerDeadlineWork(
  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
  queue: DeliveryQueue,
  task: {
-    run: () => Promise<unknown>;
+    run: (signal: AbortSignal) => Promise<unknown>;
    next: () => Promise<number | null>;
    canRun: () => boolean;
    retryMs: number;
@@ -21,7 +21,7 @@ export function registerDeadlineWork(
    retryMs: task.retryMs,
    pausePassiveMaintenance: true,
    onSettled: () => { requested++; },
-    run: async () => {
+    run: async signal => {
      if (!task.canRun()) { requested++; return; }
      if (Date.now() < earliest) { next = earliest; return; }
      if (Date.now() < coalesceUntil && (next === null || next > Date.now())) {
@@ -30,7 +30,8 @@ export function registerDeadlineWork(
      }
      const generation = ++requested;
      earliest = Date.now() + task.retryMs;
-      await task.run();
+      await task.run(signal);
+      if (signal.aborted) return;
      const deadline = await task.next();
      // Back off failed or still-overdue work, not a successfully read future deadline.
      if (deadline === null || deadline > Date.now()) earliest = 0;
@@ -0,0 +1,20 @@
+import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
+import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
+import { registerDeadlineWork } from "./deadline-work.js";
+import type { decisionService } from "./decisions.js";
+
+export function registerDecisionContinuationWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: Pick<ReturnType<typeof decisionService>, "sweepContinuations">,
+  canRun: () => boolean,
+) {
+  let next: number | null = null;
+  return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionContinuation, {
+    retryMs: 30_000, notificationCoalesceMs: 30_000, canRun,
+    run: async signal => {
+      const result = await service.sweepContinuations(100, signal);
+      next = result.retry ? Date.now() + 30_000 : null;
+    },
+    next: async () => next,
+  });
+}
@@ -0,0 +1,66 @@
+import { and, asc, eq, gt, sql } from "drizzle-orm";
+import { databaseWorkScopeKey, decisions, type Db } from "@paperclipai/db";
+
+export type DecisionContinuationWake = (input: {
+  companyId: string; agentId: string; issueId: string; decisionId: string;
+  outcome: "decided" | "expired" | "cancelled";
+}) => Promise<unknown>;
+
+const flights = new WeakMap<object, Map<string, Promise<void>>>();
+const ready = sql`${decisions.continuationPolicy} = 'wake_origin_agent'
+  AND ${decisions.metadata} ->> 'continuationPending' = 'true'
+  AND (${decisions.status} IN ('expired', 'cancelled') OR
+    (${decisions.status} = 'decided' AND ${decisions.executionStatus} IN ('succeeded', 'partial', 'failed')))`;
+
+export function decisionContinuationService(db: Db, wake: DecisionContinuationWake) {
+  const scope = databaseWorkScopeKey(db);
+  let pending = flights.get(scope);
+  if (!pending) flights.set(scope, pending = new Map());
+  const inFlight = pending;
+  let recoveryCursor: string | undefined;
+
+  function deliver(id: string): Promise<void> {
+    const existing = inFlight.get(id);
+    if (existing) return existing;
+    const attempt = (async () => {
+      const [decision] = await db.select().from(decisions).where(and(eq(decisions.id, id), ready));
+      if (!decision) return;
+      await wake({ companyId: decision.companyId, agentId: decision.originAgentId,
+        issueId: decision.originIssueId, decisionId: id,
+        outcome: decision.status === "expired" ? "expired" : decision.status === "cancelled" ? "cancelled" : "decided" });
+      await db.update(decisions).set({
+        metadata: sql`coalesce(${decisions.metadata}, '{}'::jsonb) || jsonb_build_object(
+          'continuationPending', false, 'continuationDeliveredAt', ${new Date().toISOString()}::text)`,
+        updatedAt: new Date(),
+      }).where(and(eq(decisions.id, id), ready));
+    })().finally(() => { inFlight.delete(id); });
+    inFlight.set(id, attempt);
+    return attempt;
+  }
+
+  async function sweep(maxRows = 100, signal?: AbortSignal) {
+    let attempted = 0;
+    const errors: unknown[] = [];
+    while (attempted < maxRows) {
+      if (signal?.aborted) return { retry: true };
+      const limit = Math.min(100, maxRows - attempted);
+      const rows = await db.select({ id: decisions.id }).from(decisions)
+        .where(and(ready, recoveryCursor ? gt(decisions.id, recoveryCursor) : undefined))
+        .orderBy(asc(decisions.id)).limit(limit);
+      for (const row of rows) {
+        if (signal?.aborted) return { retry: true };
+        try { await deliver(row.id); } catch (error) { errors.push(error); }
+        recoveryCursor = row.id;
+      }
+      attempted += rows.length;
+      if (rows.length < limit) { recoveryCursor = undefined; break; }
+    }
+    if (errors.length === 1) throw errors[0];
+    if (errors.length) throw new AggregateError(errors, "Decision continuation delivery failed");
+    if (signal?.aborted) return { retry: true };
+    const remaining = await db.select({ id: decisions.id }).from(decisions).where(ready).limit(1);
+    return { retry: remaining.length > 0 };
+  }
+
+  return { deliver, sweep };
+}
@@ -0,0 +1,9 @@
+import type { Db } from "@paperclipai/db";
+import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
+
+export function writeDecisionChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
+  return db.transaction(async transaction => {
+    await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation);
+    return write(transaction as unknown as Db);
+  });
+}
@@ -1,7 +1,7 @@
 import { randomUUID } from "node:crypto";
-import { and, asc, count, desc, eq, gt, gte, inArray, lte, or, sql } from "drizzle-orm";
+import { and, asc, count, desc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
-import { companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db";
+import { databaseWorkScopeKey, companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db";
 import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared";
 import type { AttentionArchiveManifestEntry, DecisionEffect, DecisionInput, DecisionOption, DecisionStatsCounts, DecisionStatsResponse } from "@paperclipai/shared";
 import { conflict, forbidden, notFound, tooManyRequests, unprocessable } from "../errors.js";
@@ -14,11 +14,16 @@ import {
 } from "./issues.js";
 import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js";

+import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js";
+import { writeDecisionChange } from "./decision-work-notifications.js";
+
 type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; updatedAt: string;
  descendantCount?: number; descendantIds?: string[]; childCount?: number; attentionArchive?: unknown };
-type Wake = (input: { companyId: string; agentId: string; issueId: string; decisionId: string; outcome: "decided" | "expired" | "cancelled" }) => Promise<unknown>;
-export type DecisionServiceOptions = { wakeOriginAgent: Wake };
+
+export type DecisionServiceOptions = { wakeOriginAgent: DecisionContinuationWake };
 const DAY = 86_400_000;
+type DecisionOutcome = typeof decisions.$inferSelect & { executions: Array<typeof decisionEffectExecutions.$inferSelect> };
+const effectFlights = new WeakMap<object, Map<string, Promise<DecisionOutcome>>>();

 function targetIds(options: DecisionOption[]) {
  const result = new Set<string>();
@@ -118,6 +123,11 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) {
 }

 export function decisionService(db: Db, options: DecisionServiceOptions) {
+  const continuations = decisionContinuationService(db, options.wakeOriginAgent);
+  const scope = databaseWorkScopeKey(db);
+  let flights = effectFlights.get(scope);
+  if (!flights) effectFlights.set(scope, flights = new Map());
+  const executing = flights;
  const authz = authorizationService(db);
  let targetSweepCursor: string | null = null;
  type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null;
@@ -544,7 +554,19 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
    }
  }

-  async function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
+  function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor): Promise<DecisionOutcome> {
+    const existing = executing.get(decision.id);
+    if (existing) return existing;
+    const attempt = (async () => {
+      const current = await get(decision.id);
+      if (!current) throw notFound("Decision not found");
+      return current.executionStatus === "running" ? executeDecisionEffects(current, userActor) : outcome(current.id);
+    })().finally(() => { executing.delete(decision.id); });
+    executing.set(decision.id, attempt);
+    return attempt;
+  }
+
+  async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
    const option = decision.options.find((item) => item.id === decision.chosenOptionId);
    if (!option || !decision.decidedByUserId) throw unprocessable("Stored decision outcome is invalid");
    const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null);
@@ -555,8 +577,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
        options: decision.options,
        targetSnapshots: decision.targetSnapshots as Record<string, Snapshot>,
      }), decision.signedSpec)) {
-        await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature" } })
-          .where(eq(decisions.id, decision.id));
+        await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
+          .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
        return outcome(decision.id);
      }
      if (option.id === "archive") {
@@ -566,8 +588,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
        );
        const manifestHash = manifest ? hashAttentionArchiveManifest(manifest) : null;
        if (!manifest || manifestHash !== metadata.manifestHash) {
-          await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch" } })
-            .where(eq(decisions.id, decision.id));
+          await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
+            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
          return outcome(decision.id);
        }
        const originActor: AuthorizationActor = {
@@ -588,50 +610,30 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
          });
        } catch (error) {
          const message = error instanceof Error ? error.message : "archive_proposal_failed";
-          await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message } })
-            .where(eq(decisions.id, decision.id));
+          await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
+            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
          return outcome(decision.id);
        }
      }
-      await db.update(decisions).set({
+      await writeDecisionChange(db, async tx => tx.update(decisions).set({
        executionStatus: "succeeded",
        updatedAt: new Date(),
        metadata: { ...metadata, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) },
-      }).where(eq(decisions.id, decision.id));
+      }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
      return outcome(decision.id);
    }
    for (let index = 0; index < option.effects.length; index += 1) await executeEffect(decision, option.effects[index]!, index, userActor, decision.decidedByUserId, run?.responsibleUserId ?? null);
    const rows = await db.select().from(decisionEffectExecutions).where(eq(decisionEffectExecutions.decisionId, decision.id));
    const successful = rows.filter((row) => row.status === "executed").length;
    const status = rows.every((row) => row.status === "executed") ? "succeeded" : successful ? "partial" : "failed";
-    await db.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata,
-      ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(eq(decisions.id, decision.id));
+    await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata,
+      ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
    return outcome(decision.id);
  }

-  async function deliverContinuation(
-    decision: typeof decisions.$inferSelect,
-    outcome: "decided" | "expired" | "cancelled",
-  ) {
-    const metadata = decision.metadata as Record<string, unknown>;
-    if (decision.continuationPolicy !== "wake_origin_agent" || metadata.continuationPending !== true) return;
-    // Delivery is intentionally at-least-once: a crash after wakeup but before
-    // this acknowledgement can retry, while a crash before wakeup cannot lose
-    // the continuation. Heartbeat wakeups coalesce concurrent queued work.
-    await options.wakeOriginAgent({ companyId: decision.companyId, agentId: decision.originAgentId,
-      issueId: decision.originIssueId, decisionId: decision.id, outcome });
-    await db.update(decisions).set({
-      metadata: sql`coalesce(${decisions.metadata}, '{}'::jsonb) || jsonb_build_object(
-        'continuationPending', false,
-        'continuationDeliveredAt', ${new Date().toISOString()}::text
-      )`,
-      updatedAt: new Date(),
-    }).where(and(eq(decisions.id, decision.id), sql`${decisions.metadata} ->> 'continuationPending' = 'true'`));
-  }
-
  async function resumeDecision(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
    const result = decision.executionStatus === "running" ? await runEffects(decision, userActor) : await outcome(decision.id);
-    await deliverContinuation(result, "decided");
+    await continuations.deliver(result.id);
    return outcome(decision.id);
  }

@@ -651,25 +653,25 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
    }
    if (current.status !== "open") throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
    if (current.expiresAt <= new Date()) {
-      const [expired] = await db.update(decisions).set({ status: "expired", updatedAt: new Date(), metadata: { ...metadata, expiredReason: "ttl",
+      const [expired] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: new Date(), metadata: { ...metadata, expiredReason: "ttl",
        ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } })
-        .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"), lte(decisions.expiresAt, new Date()))).returning();
+        .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"), lte(decisions.expiresAt, new Date()))).returning());
      if (expired) {
        await logActivity(db, { companyId: expired.companyId, actorType: "system", actorId: "decision-expiry-sweeper", agentId: expired.originAgentId,
          runId: expired.originRunId, action: "decision.expired", entityType: "decision", entityId: expired.id, details: { expiredReason: "ttl" } });
-        await deliverContinuation(expired, "expired");
+        await continuations.deliver(expired.id);
      }
      throw conflict("decision_expired", { code: "decision_expired" });
    }
    if (!current.options.some((option) => option.id === input.optionId)) throw unprocessable("Unknown optionId");
    const values = input.inputValues ?? {};
    for (const field of current.inputs ?? []) { const value = values[field.id] ?? ""; if (field.required && !value.trim()) throw unprocessable(`Input ${field.id} is required`); if (field.maxLength && value.length > field.maxLength) throw unprocessable(`Input ${field.id} is too long`); }
-    const [claimed] = await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values,
+    const [claimed] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values,
      decidedByUserId: input.decidedByUserId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...metadata,
        decideIdempotencyKey: input.idempotencyKey ?? null,
        ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}),
        ...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}) } })
-      .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning();
+      .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning());
    if (!claimed) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
    const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, claimed.originRunId)).then((rows) => rows[0] ?? null);
    await logActivity(db, { companyId: claimed.companyId, actorType: "system", actorId: "decision-executor", agentId: claimed.originAgentId, runId: claimed.originRunId,
@@ -683,14 +685,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
    const current = await get(id); if (!current) throw notFound("Decision not found");
    if (actor.actorType === "agent" && actor.actorId !== current.originAgentId) throw forbidden("Only the origin agent may cancel");
    if (current.status === "cancelled") {
-      await deliverContinuation(current, "cancelled");
+      await continuations.deliver(current.id);
      return (await get(id))!;
    }
-    const [updated] = await db.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata,
-      ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning();
+    const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata,
+      ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning());
    if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
    await logActivity(db, { companyId: updated.companyId, actorType: actor.actorType, actorId: actor.actorId, runId: actor.runId, action: "decision.cancelled", entityType: "decision", entityId: id });
-    await deliverContinuation(updated, "cancelled");
+    await continuations.deliver(updated.id);
    return (await get(id))!;
  }

@@ -699,14 +701,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
    if (!verifyDecisionSpec(spec({ id: current.id, options: current.options, targetSnapshots: current.targetSnapshots as Record<string, Snapshot> }), current.signedSpec)) throw forbidden("Decision signature verification failed");
    const empty = current.options.find((option) => option.effects.length === 0);
    if (empty) return decide({ id, optionId: empty.id, decidedByUserId: userId, userActor, dismissed: true, dismissReason: reason });
-    const [updated] = await db.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId,
+    const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId,
      decidedAt: new Date(), updatedAt: new Date(), metadata: { ...current.metadata, dismissed: true, dismissReason: reason ?? null,
-        ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning();
+        ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning());
    if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" });
    await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-executor", agentId: updated.originAgentId,
      runId: updated.originRunId, responsibleUserIdOverride: userId, action: "decision.dismissed", entityType: "decision", entityId: updated.id,
      details: { chosenOptionId: "dismissed", decidedByUserId: userId, dismissed: true } });
-    await deliverContinuation(updated, "decided");
+    await continuations.deliver(updated.id);
    return outcome(id);
  }

@@ -718,7 +720,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
      return { ...bundle, decisions: created }; });
  }

-  async function sweepExpired(now = new Date()) {
+  async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean } = {}) {
    const configuredBatchSize = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100);
    const batchSize = Number.isFinite(configuredBatchSize)
      ? Math.max(1, Math.trunc(configuredBatchSize))
@@ -737,17 +739,9 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
      const recovered = await runEffects(decision, await recoveryActor(decision.companyId, decision.decidedByUserId));
      if (recovered.executionStatus === "running") continue;
      resumed += 1;
-      await deliverContinuation(recovered, "decided");
-    }
-    const pendingContinuations = await db.select().from(decisions)
-      .where(and(eq(decisions.continuationPolicy, "wake_origin_agent"),
-        sql`${decisions.metadata} ->> 'continuationPending' = 'true'`,
-        or(inArray(decisions.status, ["expired", "cancelled"]), and(eq(decisions.status, "decided"),
-          inArray(decisions.executionStatus, ["succeeded", "partial", "failed"])))))
-      .orderBy(asc(decisions.updatedAt)).limit(batchSize);
-    for (const decision of pendingContinuations) {
-      await deliverContinuation(decision, decision.status === "expired" ? "expired" : decision.status === "cancelled" ? "cancelled" : "decided");
+      await continuations.deliver(recovered.id);
    }
+    if (options.includeContinuations !== false) await continuations.sweep(batchSize);
    const ttlRows = await db.select().from(decisions)
      .where(and(eq(decisions.status, "open"), lte(decisions.expiresAt, now)))
      .orderBy(asc(decisions.expiresAt)).limit(batchSize);
@@ -774,13 +768,13 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
        targets.some((target) => snapshots[target.id]?.status !== "done");
      if (!targetGone && !targetsCompleted && decision.expiresAt >= now) continue;
      const reason = targetGone ? "target_gone" : targetsCompleted ? "target_completed" : "ttl";
-      const [updated] = await db.update(decisions).set({ status: "expired", updatedAt: now, metadata: { ...decision.metadata, expiredReason: reason,
-        ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.status, "open"))).returning();
+      const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: now, metadata: { ...decision.metadata, expiredReason: reason,
+        ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.status, "open"))).returning());
      if (!updated) continue; expired += 1;
      await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-expiry-sweeper", agentId: updated.originAgentId, runId: updated.originRunId, action: "decision.expired", entityType: "decision", entityId: updated.id, details: { expiredReason: reason } });
-      await deliverContinuation(updated, "expired"); }
+      await continuations.deliver(updated.id); }
    return { expired, resumed };
  }

-  return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired };
+  return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep };
 }
@@ -9,6 +9,8 @@ export const DELIVERY_QUEUES = {
  executionStatus: "execution-status",
  executionDisposition: "execution-disposition",
  environmentCleanup: "environment-cleanup",
+  decisionContinuation: "decision-continuation",
+
  decisionArchiveDelivery: "decision-archive-delivery",

  toolReviewProjection: "tool-review-projection",
				`@@ -0,0 +1 @@`
				`CREATE INDEX CONCURRENTLY IF NOT EXISTS "decisions_pending_continuation_idx" ON "decisions" USING btree ("id") WHERE "decisions"."continuation_policy" = 'wake_origin_agent' AND "decisions"."metadata" ->> 'continuationPending' = 'true' AND ("decisions"."status" IN ('expired', 'cancelled') OR ("decisions"."status" = 'decided' AND "decisions"."execution_status" IN ('succeeded', 'partial', 'failed')));`