fix(server): wake decision continuations from committed changes

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

doc/idle-decision-continuations.md

@@ -0,0 +1,106 @@
+# 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.
+It reads at most 100 IDs per page and advances past failed candidates before
+reporting failure. 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. 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 admitted delivery. 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 92 tests: 49 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.
+
+Migration 0382 applies and replays without error. 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.

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -15,7 +15,7 @@ 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), all checks passed, Greptile 5/5 at `36844a458b` |
 | 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: [#17475](https://github.com/paperclipai/paperclip/pull/17475), local verification passed, Greptile 5/5 at `692e3ee02c`, CI running |
+| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; notification delivery: [#17475](https://github.com/paperclipai/paperclip/pull/17475), all checks passed (Paperclip Review skipped), Greptile 5/5 at `692e3ee02c` |
 | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending |
 | 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), checks passed at `d41590ac0e`; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), all checks passed, Greptile 5/5 at `c5c53fc43f`; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), all checks passed, Greptile 5/5 at `ce6a260a83`; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), all checks passed, Greptile 5/5 at `7a1cf03e62`; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), all checks passed, Greptile 5/5 at `85003a7ac4`; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), all checks passed, Greptile 5/5 at `0956883a84` |
@@ -369,3 +369,26 @@ 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.

packages/db/src/client.ts

@@ -507,6 +507,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
   "0378_mean_scream.sql",
   "0379_narrow_the_executioner.sql",
   "0381_open_madame_web.sql",
+  "0382_chemical_madame_web.sql",
   "0380_funny_adam_warlock.sql",
 ]);
 

packages/db/src/index.ts

@@ -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";

packages/db/src/migrations/0382_chemical_madame_web.sql

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

packages/db/src/migrations/meta/0382_snapshot.json

@@ -1,6 +1,6 @@
 {
-  "id": "a0fcd9b8-f125-4985-8795-f567466b9aef",
-  "prevId": "7832df38-fa1b-4687-bd06-eccbc29bd28c",
+  "id": "5e8d042e-d697-4753-9582-7d0c042ef893",
+  "prevId": "afdc5e70-19b3-416d-917a-beb9fdc1e2e1",
   "version": "7",
   "dialect": "postgresql",
   "tables": {
@@ -24030,6 +24030,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": [
@@ -52787,6 +52803,66 @@
           "method": "btree",
           "with": {}
         },
+        "tool_action_requests_pending_expiry_idx": {
+          "name": "tool_action_requests_pending_expiry_idx",
+          "columns": [
+            {
+              "expression": "expires_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"tool_action_requests\".\"status\" = 'pending'",
+          "concurrently": false,
+          "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": [
@@ -55197,6 +55273,21 @@
           "method": "btree",
           "with": {}
         },
+        "tool_gateway_sessions_expiry_idx": {
+          "name": "tool_gateway_sessions_expiry_idx",
+          "columns": [
+            {
+              "expression": "expires_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
         "tool_gateway_sessions_run_idx": {
           "name": "tool_gateway_sessions_run_idx",
           "columns": [

packages/db/src/migrations/meta/_journal.json

@@ -2661,6 +2661,13 @@
       "when": 1792863626931,
       "tag": "0381_open_madame_web",
       "breakpoints": true
+    },
+    {
+      "idx": 382,
+      "version": "7",
+      "when": 1792871748836,
+      "tag": "0382_chemical_madame_web",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/decisions.ts

@@ -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")

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

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

packages/db/src/work-signals.ts

@@ -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;
+}

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

@@ -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,
@@ -26,7 +27,15 @@ import { decisionService } from "../services/decisions.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 +50,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 +80,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" }] });
@@ -656,4 +668,210 @@ 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() {
+      coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining,
+        canReconcile: () => true, onError: error => errors.push(error) });
+      await registerDecisionContinuationWork(coordinator, workerService(), () => !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.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(); await markReady(row.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;
+      await delivered(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"); });
+      await start();
+      expect(wake).toHaveBeenCalledTimes(101); await delivered(ids.at(-1)!); await pending(ids[0]);
+      expect(errors).toHaveLength(1); errors.length = 0; wake.mockImplementation(async () => {});
+      await quiet(); await delivered(ids[0]); expect(wake).toHaveBeenCalledTimes(102);
+    });
+  });
+
 });

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

@@ -5,7 +5,7 @@ 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 { reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js";
+import { decisionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js";
 import { runDatabaseBackup } 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";
@@ -340,6 +340,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),
@@ -855,6 +856,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();
     }
@@ -891,6 +896,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();
     }

server/src/index.ts

@@ -1,3 +1,4 @@
+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";
 import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js";
@@ -1800,7 +1801,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 },
@@ -1821,7 +1824,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) => {

server/src/services/decision-continuation-work.ts

@@ -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, canRun,
+    run: async () => {
+      const result = await service.sweepContinuations();
+      next = result.retry ? Date.now() + 30_000 : null;
+    },
+    next: async () => next,
+  });
+}

server/src/services/decision-continuations.ts

@@ -0,0 +1,63 @@
+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;
+
+  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 = Infinity) {
+    let cursor: string | undefined;
+    let attempted = 0;
+    const errors: unknown[] = [];
+    while (attempted < maxRows) {
+      const limit = Math.min(100, maxRows - attempted);
+      const rows = await db.select({ id: decisions.id }).from(decisions)
+        .where(and(ready, cursor ? gt(decisions.id, cursor) : undefined))
+        .orderBy(asc(decisions.id)).limit(limit);
+      for (const row of rows) {
+        try { await deliver(row.id); } catch (error) { errors.push(error); }
+      }
+      attempted += rows.length;
+      if (rows.length < limit) break;
+      cursor = rows.at(-1)!.id;
+    }
+    if (errors.length === 1) throw errors[0];
+    if (errors.length) throw new AggregateError(errors, "Decision continuation delivery failed");
+    const remaining = await db.select({ id: decisions.id }).from(decisions).where(ready).limit(1);
+    return { retry: remaining.length > 0 };
+  }
+
+  return { deliver, sweep };
+}

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

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

server/src/services/decisions.ts

@@ -1,5 +1,5 @@
 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 { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared";
@@ -14,10 +14,13 @@ 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;
 
 function targetIds(options: DecisionOption[]) {
@@ -118,6 +121,7 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) {
 }
 
 export function decisionService(db: Db, options: DecisionServiceOptions) {
+  const continuations = decisionContinuationService(db, options.wakeOriginAgent);
   const authz = authorizationService(db);
   let targetSweepCursor: string | null = null;
   type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null;
@@ -555,8 +559,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" } })
+          .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
         return outcome(decision.id);
       }
       if (option.id === "archive") {
@@ -566,8 +570,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" } })
+            .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running"))));
           return outcome(decision.id);
         }
         const originActor: AuthorizationActor = {
@@ -588,50 +592,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 } })
+            .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 +635,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 +667,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 +683,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 +702,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 +721,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 +750,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 };
 }

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

@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
   executionStatus: "execution-status",
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
+  decisionContinuation: "decision-continuation",
   decisionArchiveDelivery: "decision-archive-delivery",
   toolReviewProjection: "tool-review-projection",
   toolReviewInterruption: "tool-review-interruption",