fix(server): recover decision effects from saved claim deadlines

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

doc/idle-decision-effect-recovery.md

@@ -0,0 +1,97 @@
+# Idle decision-effect recovery
+
+Interrupted decision effects now have a dedicated deadline worker. The main
+maintenance tick no longer scans running decisions. Target invalidation remains
+in that tick; TTL expiry and continuation delivery have their own workers.
+
+## Obligation and activation
+
+The existing decision row is durable truth. A row with `status = decided` and
+`executionStatus = running` owes effect recovery. Its next deadline is
+`updatedAt + PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS`, default 60 seconds. Zero is
+supported; invalid or negative values retain the existing default. Reads do not
+move the timestamp. Open decisions and completed history create no recovery debt.
+
+The existing `notifyDecisionChange` helper adds the recovery topic. Creation,
+bundle creation, foreground running claims, effect completion, archive completion
+and failure, cancellation, dismissal, expiry, and continuation acknowledgements
+already pass through that boundary. Notifications are awaited before writes in
+the existing transaction. Outer commit activates work; rollback does not publish
+a claim. The additional topic does not wrap provider work or agent wakes in SQL
+transactions and does not introduce a new queue table.
+
+The supported writer model is the application DB signal owner and its child pools.
+Independent SQL clients and multiple active writers need a separate notification
+contract. Startup reconstructs committed claims; it does not observe future writes
+from unsupported writers.
+
+## Ownership and recovery
+
+`decision-effect-recovery.ts` owns selection, grace, bounded paging, and earliest
+deadline lookup. `decisions.ts` owns execution, current membership, effect claims,
+audit, archive policy, and continuation delivery. The worker adapter only binds
+that domain contract to the shared coordinator. Startup registers it only when
+heartbeat scheduling is enabled. The public combined sweep retains its default
+recovery behavior for existing callers.
+
+Each pass honors the existing configured batch size (default 100, finite values
+clamped to at least one integer). Its timestamp/ID cursor advances past failed or
+invalid candidates and wraps at the end. Failed rows retain their obligation.
+Rows without a saved decider are not executed under invented authority. A missing
+option produces an error. Both cases remain visible as unfinished work and retain
+retry scheduling rather than starving later candidates or being discarded.
+
+The decision service re-reads current status before recovery. Foreground and
+recovery execution share the existing per-decision flight scoped to the DB owner.
+Individual effects keep their durable records and transaction advisory locks.
+Completed effects are not repeated after a failed final decision status write.
+Recovery derives authority from the saved decider's current active membership.
+It preserves the current authorization intersection and current target checks.
+
+The scheduler has a 30-second minimum retry spacing for pending or failed work.
+A first claim or failed retry during drain retains its wake without passive SQL.
+Shutdown stops subscriptions and checks cancellation between candidates. An
+already admitted effect/continuation is awaited; later candidates remain durable
+for restart. No periodic pass remains after all claims settle.
+
+## Existing guarantees and limits
+
+Regular effects preserve their existing per-effect transaction boundaries.
+Continuation delivery retains its separate at-least-once wake/acknowledgement
+contract. This migration does not establish exactly-once external delivery.
+
+Archive proposals have a separate versioned manifest contract. Recovery still
+verifies the saved signature/manifest and current authority. A crash after an
+archive commits but before the decision settles can encounter the changed
+manifest version on restart. The existing result is a failed decision with an
+archive error, while the archive remains committed once. The new worker preserves
+and tests that outcome; it does not claim to repair this preexisting atomicity gap.
+
+Cloud sleep guards remain conservative. The internal next deadline is not a
+persisted external alarm and does not authorize host sleep with running decisions.
+This change establishes worker quiet, not actual instance/database suspension.
+
+## Verification
+
+The new real-PostgreSQL suite covers 17 cases: empty/open/completed-history quiet,
+real foreground claim and failed settlement, exact grace and zero override,
+restart from a claimed effect, active foreground/recovery overlap, outer commit
+and rollback, first claim and failed retry through drain, revoked authority,
+continuation failure, aggregate exclusion, fair paging past missing-decider and
+malformed claims, archive recovery before and after an archive commit, a claim
+created during delivery, and shutdown between candidates. Quiet assertions observe
+SQL methods through one hour of fake time after settling.
+
+The combined selection passes 212 tests: the 17 new cases, 50 existing decision
+service cases, 14 TTL cases, 39 startup cases, and 92 agent-history route cases.
+The first 15-case recovery run also passed. Workspace typecheck/build and the
+final server typecheck pass. No failure was suppressed or timeout increased.
+
+Migration 0384 adds one partial index concurrently, with `IF NOT EXISTS`, through
+the migration owner's top-level statement path. Generation retains the newest
+five schema snapshots. Fresh application, replay, safety, and numbering checks
+pass. With 40,000 completed decisions and one running claim, both due selection
+and earliest-deadline lookup use the partial index and read two buffers (0.012 ms
+and 0.003 ms in the local probe). The probe transaction is rolled back.
+
+Full PR CI and review must pass separately before this slice is ready.

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

@@ -16,7 +16,7 @@ Do not mark an item done from a timer removal or a green check alone.
 | 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), all checks passed (Paperclip Review skipped), Greptile 5/5 at `692e3ee02c` |
-| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations: [#17478](https://github.com/paperclipai/paperclip/pull/17478), Greptile 5/5 at `5f96892a58`, CI running; TTL: [#17479](https://github.com/paperclipai/paperclip/pull/17479), local verification passed at `63b58cc26f`, CI/review running; effects and target invalidation pending |
+| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations: [#17478](https://github.com/paperclipai/paperclip/pull/17478), all checks passed, Greptile 5/5 at `5f96892a58`; TTL: [#17479](https://github.com/paperclipai/paperclip/pull/17479), Greptile 5/5 at `63b58cc26f`, one CI timeout retry running; effects and target invalidation 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` |
 | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending |
@@ -490,3 +490,25 @@ archive failure semantics, outer commit/rollback, failed retry through drain,
 changes during execution, fair bounded paging, and shutdown. Run existing
 archive and continuation regressions on the combined stack. The full original
 inventory remains the goal after this slice.
+
+
+## Slice 11: interrupted decision effects
+
+The effect-recovery owner and worker now use committed decision changes and the
+minimum running claim timestamp plus the existing recovery grace. The aggregate
+tick opts out of this duty while preserving target invalidation. Effect execution,
+current authority, claims, audit, and archive policy remain with the decision
+service. See [effect recovery](../idle-decision-effect-recovery.md).
+
+All 212 selected tests pass: 17 new recovery cases, 50 decision service cases,
+14 TTL cases, 39 startup cases, and 92 agent-history route cases. Workspace
+typecheck/build and final server typecheck pass. Migration apply/replay and a
+40,000-row completed-history query probe pass; both lookups use the new partial
+index with two buffers. PR creation and full CI/review are next.
+
+The continuation parent #17478 now has passing checks and Greptile 5/5 at
+`5f96892a58`, with both review findings resolved. TTL #17479 has Greptile 5/5 at
+`63b58cc26f` and no findings. Its first CI run timed out in the unchanged
+agent-history route test. All 92 route tests passed locally; one failed-job retry
+was requested. This evidence does not yet establish green TTL CI. The full
+original inventory remains in scope; no stack PR has been merged.

packages/db/src/client.ts

@@ -509,6 +509,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
   "0381_open_madame_web.sql",
   "0382_chemical_madame_web.sql",
   "0383_clever_chamber.sql",
+  "0384_many_reaper.sql",
   "0380_funny_adam_warlock.sql",
 ]);
 

packages/db/src/migrations/0384_many_reaper.sql

@@ -0,0 +1 @@
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "decisions_running_recovery_idx" ON "decisions" USING btree ("updated_at","id") WHERE "decisions"."status" = 'decided' AND "decisions"."execution_status" = 'running';

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

@@ -1,6 +1,6 @@
 {
-  "id": "b8e3cf47-6bf6-487f-adee-5642c0f97ea8",
-  "prevId": "01c4455a-e59c-43e6-940b-09b9c76171d8",
+  "id": "45d3a35e-2d6d-4fc1-8445-c35eef1ea173",
+  "prevId": "75e7a97b-132d-4326-b683-003c1a2b7048",
   "version": "7",
   "dialect": "postgresql",
   "tables": {
@@ -24030,6 +24030,66 @@
           "method": "btree",
           "with": {}
         },
+        "decisions_running_recovery_idx": {
+          "name": "decisions_running_recovery_idx",
+          "columns": [
+            {
+              "expression": "updated_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'",
+          "concurrently": false,
+          "method": "btree",
+          "with": {}
+        },
+        "decisions_open_expiry_idx": {
+          "name": "decisions_open_expiry_idx",
+          "columns": [
+            {
+              "expression": "expires_at",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            },
+            {
+              "expression": "id",
+              "isExpression": false,
+              "asc": true,
+              "nulls": "last"
+            }
+          ],
+          "isUnique": false,
+          "where": "\"decisions\".\"status\" = 'open'",
+          "concurrently": false,
+          "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": [
@@ -52809,6 +52869,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": [

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

@@ -2675,6 +2675,13 @@
       "when": 1792872323806,
       "tag": "0383_clever_chamber",
       "breakpoints": true
+    },
+    {
+      "idx": 384,
+      "version": "7",
+      "when": 1792873408795,
+      "tag": "0384_many_reaper",
+      "breakpoints": true
     }
   ]
 }

packages/db/src/schema/decisions.ts

@@ -69,6 +69,8 @@ export const decisions = pgTable(
       table.status,
       table.expiresAt,
     ),
+    runningRecoveryIdx: index("decisions_running_recovery_idx").on(table.updatedAt, table.id)
+      .where(sql`${table.status} = 'decided' AND ${table.executionStatus} = 'running'`),
     openExpiryIdx: index("decisions_open_expiry_idx").on(table.expiresAt, table.id)
       .where(sql`${table.status} = 'open'`),
     pendingContinuationIdx: index("decisions_pending_continuation_idx").on(table.id)

server/src/__tests__/decision-effect-recovery-work.test.ts

@@ -0,0 +1,290 @@
+import { randomUUID } from "node:crypto";
+import { eq, sql } from "drizzle-orm";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { activityLog, agents, authUsers, companies, companyMemberships, createDb, decisionEffectExecutions,
+  decisions, decisionRetention, decisionTargetIssues, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
+import { decisionService } from "../services/decisions.js";
+import { registerDecisionEffectRecoveryWork } from "../services/decision-effect-recovery-work.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 { hashAttentionArchiveManifest } from "../services/decision-retention.js";
+import type { AttentionArchiveManifestEntry } from "@paperclipai/shared";
+import { idleWorkSnapshot } from "../services/task-admission.js";
+
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISION_EFFECTS_TEST_DATABASE_URL");
+const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
+(support.supported ? describe : describe.skip)("decision effect recovery worker", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: ReturnType<typeof createDb>;
+  let companyId: string, agentId: string, originId: string, targetId: string, runId: string, userId: string;
+  let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
+  let draining: boolean;
+  let recovery: ReturnType<typeof decisionService>;
+  const errors: unknown[] = [];
+  const wake = vi.fn(async () => {});
+  const service = () => decisionService(db, { wakeOriginAgent: wake });
+
+  beforeAll(async () => {
+    if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("decision-effects-");
+    db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
+  }, 30_000);
+  beforeEach(async () => {
+    vi.stubEnv("PAPERCLIP_DECISION_SIGNING_SECRET", "0123456789abcdef0123456789abcdef");
+    companyId = randomUUID(); agentId = randomUUID(); originId = randomUUID(); targetId = randomUUID(); runId = randomUUID(); userId = randomUUID();
+    await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `R${companyId.slice(0, 6)}`, requireBoardApprovalForNewAgents: false });
+    await db.insert(authUsers).values({ id: userId, name: "Decider", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" });
+    await db.insert(agents).values({ id: agentId, companyId, name: "Proposer", role: "engineer", status: "active", adapterType: "codex_local", adapterConfig: {}, runtimeConfig: {}, permissions: {} });
+    await db.insert(issues).values([
+      { id: originId, companyId, title: "Origin", status: "in_progress", priority: "medium", assigneeAgentId: agentId, responsibleUserId: userId },
+      { id: targetId, companyId, title: "Target", status: "todo", priority: "medium", responsibleUserId: userId },
+    ]);
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", responsibleUserId: userId, contextSnapshot: { issueId: originId } });
+    await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
+    draining = false; errors.length = 0; wake.mockReset();
+    vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
+  });
+  afterEach(async () => {
+    await coordinator?.stop(); coordinator = undefined;
+    vi.useRealTimers(); vi.restoreAllMocks(); vi.unstubAllEnvs();
+    await db.execute(sql`drop trigger if exists decision_effect_finish_fail on decisions`);
+    await db.execute(sql`drop function if exists decision_effect_finish_fail()`);
+    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(heartbeatRuns);
+    await db.delete(issues); await db.delete(agents); await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies);
+    expect(errors).toEqual([]);
+  });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+
+  const actor = () => ({ type: "agent" as const, companyId, agentId, runId, source: "agent_jwt" as const,
+    onBehalfOfUserId: userId, onBehalfOfMemberships: [{ companyId, membershipRole: "member", status: "active" }] });
+  const board = () => ({ type: "board" as const, userId, companyIds: [companyId], source: "session" as const,
+    memberships: [{ companyId, membershipRole: "member", status: "active" }] });
+  const create = () => service().create({ companyId, agentId, runId, actor: actor(), title: "Comment?", body: "Body",
+    continuationPolicy: "wake_origin_agent", options: [{ id: "yes", label: "Yes", effects: [
+      { type: "comment_on_issue", targetIssueId: targetId, staleness: "lenient", bodyMarkdown: "hello" },
+    ] }] });
+  const decide = (id: string) => service().decide({ id, optionId: "yes", decidedByUserId: userId, userActor: board() });
+  const claim = (id: string, updatedAt = new Date(), owner = db) => writeDecisionChange(owner, async tx => tx.update(decisions).set({
+    status: "decided", executionStatus: "running", chosenOptionId: "yes", decidedByUserId: userId,
+    updatedAt, inputValues: {}, metadata: { continuationPending: true },
+  }).where(eq(decisions.id, id)));
+  async function start() {
+    coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
+      onError: error => errors.push(error) });
+    await registerDecisionContinuationWork(coordinator, service(), () => !draining).ready;
+    recovery = service();
+    vi.spyOn(recovery, "recoverEffects");
+    await registerDecisionEffectRecoveryWork(coordinator, recovery, () => !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 settled(id: string, executionStatus = "succeeded") {
+    await vi.waitFor(async () => {
+      await vi.advanceTimersByTimeAsync(30_000);
+      expect(await service().get(id)).toMatchObject({ executionStatus, metadata: { continuationPending: false } });
+    }, { timeout: 5000 });
+  }
+  const comments = () => db.select().from(issueComments).where(eq(issueComments.issueId, targetId));
+  async function failSettlement() {
+    await db.execute(sql`create function decision_effect_finish_fail() returns trigger language plpgsql as $$ begin
+      if OLD.execution_status = 'running' and NEW.execution_status <> 'running' then raise exception 'settlement unavailable'; end if;
+      return NEW; end $$`);
+    await db.execute(sql`create trigger decision_effect_finish_fail before update on decisions for each row execute function decision_effect_finish_fail()`);
+  }
+
+  it("stays quiet with open decisions and delivered history after a real decision", async () => {
+    await start(); await quiet(); const row = await create(); await quiet();
+    await decide(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
+  });
+
+  it("reactivates from the actual claim and recovers a failed settlement without repeating the effect", async () => {
+    await start(); await quiet(); const row = await create(); await failSettlement();
+    await expect(decide(row.id)).rejects.toMatchObject({ cause: { message: "settlement unavailable" } });
+    expect((await service().get(row.id))?.executionStatus).toBe("running"); expect(await comments()).toHaveLength(1);
+    await db.execute(sql`drop trigger decision_effect_finish_fail on decisions`);
+    await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
+  });
+
+  it("reconstructs the saved grace deadline at restart and recovers a claimed effect", async () => {
+    const row = await create(); const claimedAt = new Date(); await claim(row.id, claimedAt);
+    await db.insert(decisionEffectExecutions).values({ decisionId: row.id, effectIndex: 0, effectType: "comment_on_issue", targetIssueId: targetId, status: "claimed" });
+    await start(); expect(coordinator!.nextWakeAt()).toBe(claimedAt.getTime() + 60_000);
+    await vi.advanceTimersByTimeAsync(30_000); expect(await comments()).toHaveLength(0);
+    await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
+  });
+
+  it("recovers at the exact grace boundary and preserves a zero grace override", async () => {
+    const row = await create(), at = new Date(); await claim(row.id, at);
+    expect(await service().recoverEffects(new Date(at.getTime() + 59_999))).toEqual({ resumed: 0 });
+    expect(await service().recoverEffects(new Date(at.getTime() + 60_000))).toEqual({ resumed: 1 });
+    vi.stubEnv("PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS", "0");
+    const second = await create(); await claim(second.id, at);
+    expect(await service().recoverEffects(at)).toEqual({ resumed: 1 }); expect(await comments()).toHaveLength(2);
+  });
+
+  it("joins active foreground effects when their recovery deadline arrives", async () => {
+    const row = await create(); await start();
+    let release!: () => void, locked!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const acquired = new Promise<void>(resolve => { locked = resolve; });
+    const lock = db.transaction(async tx => {
+      await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`decision-effect:${row.id}:0`}, 0))`);
+      locked(); await gate;
+    });
+    await acquired;
+    const deciding = decide(row.id);
+    try {
+      await vi.waitFor(async () => expect((await service().get(row.id))?.executionStatus).toBe("running"));
+      await vi.advanceTimersByTimeAsync(120_000);
+      expect(await comments()).toHaveLength(0); expect(wake).not.toHaveBeenCalled();
+      await vi.waitFor(() => expect(recovery.recoverEffects).toHaveBeenCalledTimes(2));
+    } finally { release(); }
+    await lock; await deciding; await quiet();
+    expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
+  });
+
+  it("defers an outer transaction until commit and ignores rollback", async () => {
+    const row = await create(); await start(); await quiet();
+    await expect(db.transaction(async tx => {
+      await claim(row.id, new Date(Date.now() - 120_000), tx as unknown as typeof db);
+      await vi.advanceTimersByTimeAsync(120_000); expect(await comments()).toHaveLength(0); throw new Error("rollback");
+    })).rejects.toThrow("rollback");
+    await quiet(); expect((await service().get(row.id))?.status).toBe("open");
+    await db.transaction(async tx => {
+      await claim(row.id, new Date(Date.now() - 120_000), tx as unknown as typeof db);
+      await vi.advanceTimersByTimeAsync(120_000); expect(await comments()).toHaveLength(0);
+    });
+    await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1);
+  });
+
+  it("retains a first claim through drain without passive SQL", async () => {
+    const row = await create(); await start(); await quiet(); draining = true;
+    await claim(row.id, new Date(Date.now() - 120_000));
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+    expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    draining = false; await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1);
+  });
+
+  it("retains a failed recovery through drain and does not replay the completed effect", async () => {
+    const row = await create(); await claim(row.id, new Date(Date.now() - 120_000)); await failSettlement(); await start();
+    await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); draining = true;
+    expect(await comments()).toHaveLength(1);
+    const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(120_000);
+    expect(reads).not.toHaveBeenCalled(); reads.mockRestore();
+    await db.execute(sql`drop trigger decision_effect_finish_fail on decisions`);
+    draining = false; await settled(row.id); await quiet(); expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(1);
+    for (const error of errors) expect(error).toMatchObject({ cause: { message: "settlement unavailable" } }); errors.length = 0;
+  });
+
+  it("rechecks current membership before executing a recovered effect", async () => {
+    const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
+    await db.update(companyMemberships).set({ status: "inactive" }).where(eq(companyMemberships.principalId, userId));
+    await start(); await settled(row.id, "failed"); await quiet(); expect(await comments()).toHaveLength(0);
+    const result = await service().outcome(row.id);
+    expect(result.executions[0]?.error).toBe("deny_decision_intersection"); expect(wake).toHaveBeenCalledTimes(1);
+  });
+
+  it("leaves continuation failures to their owner after effect recovery finishes", async () => {
+    const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
+    wake.mockRejectedValueOnce(new Error("wake unavailable")); await start(); await settled(row.id); await quiet();
+    expect(await comments()).toHaveLength(1); expect(wake).toHaveBeenCalledTimes(2);
+    expect(errors.every(error => String(error).includes("wake unavailable"))).toBe(true); errors.length = 0;
+  });
+
+  it("keeps aggregate maintenance from recovering effects when the worker owns them", async () => {
+    const row = await create(); await claim(row.id, new Date(Date.now() - 120_000));
+    expect(await service().sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false }))
+      .toEqual({ expired: 0, resumed: 0 });
+    expect(await comments()).toHaveLength(0);
+    expect(await service().sweepExpired()).toEqual({ expired: 0, resumed: 1 }); expect(await comments()).toHaveLength(1);
+  });
+
+  it("pages past invalid legacy claims without inventing authority or losing their debt", async () => {
+    vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1");
+    const invalid = await create(), valid = await create();
+    await claim(invalid.id, new Date(Date.now() - 180_000)); await claim(valid.id, new Date(Date.now() - 120_000));
+    await db.update(decisions).set({ decidedByUserId: null }).where(eq(decisions.id, invalid.id));
+    const recovery = service();
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 0 }); expect(await recovery.nextEffectRecoveryAt()).not.toBeNull();
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 1 }); expect(await comments()).toHaveLength(1);
+    expect((await recovery.get(invalid.id))?.executionStatus).toBe("running");
+    expect(await recovery.nextEffectRecoveryAt()).not.toBeNull();
+  });
+
+  it("advances past malformed claims and retries them on a later pass", async () => {
+    vi.stubEnv("PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE", "1");
+    const invalid = await create(), valid = await create();
+    await claim(invalid.id, new Date(Date.now() - 180_000)); await claim(valid.id, new Date(Date.now() - 120_000));
+    await db.update(decisions).set({ chosenOptionId: "missing" }).where(eq(decisions.id, invalid.id));
+    const recovery = service();
+    await expect(recovery.recoverEffects()).rejects.toThrow("Stored decision outcome is invalid");
+    expect(await recovery.recoverEffects()).toEqual({ resumed: 1 }); expect(await comments()).toHaveLength(1);
+    await recovery.recoverEffects();
+    await expect(recovery.recoverEffects()).rejects.toThrow("Stored decision outcome is invalid");
+  });
+
+  it.each([false, true])("preserves archive recovery semantics with prior committed archive: %s", async committed => {
+    await db.update(issues).set({ status: "in_review", assigneeUserId: userId }).where(eq(issues.id, targetId));
+    const activityAt = new Date("2026-04-01T00:00:00.000Z");
+    await db.insert(decisionRetention).values({ companyId, sourceKind: "review", sourceId: targetId, sourceActivityAt: activityAt });
+    const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetId,
+      expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }];
+    const row = await service().create({ companyId, actor: actor(), 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:${targetId}`]: { status: "attention", assigneeAgentId: null,
+        assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } });
+    if (committed) {
+      await failSettlement();
+      await expect(service().decide({ id: row.id, optionId: "archive", decidedByUserId: userId, userActor: board() }))
+        .rejects.toMatchObject({ cause: { message: "settlement unavailable" } });
+      await db.execute(sql`drop trigger decision_effect_finish_fail on decisions`);
+    } else {
+      await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running",
+        chosenOptionId: "archive", decidedByUserId: userId, updatedAt: new Date(Date.now() - 120_000),
+        metadata: { ...row.metadata, continuationPending: true } }).where(eq(decisions.id, row.id)));
+    }
+    await start(); await settled(row.id, committed ? "failed" : "succeeded"); await quiet();
+    const [retained] = await db.select().from(decisionRetention);
+    expect(retained.archiveVersion).toBe(1); expect(retained.archivedAt).not.toBeNull();
+    expect(wake).toHaveBeenCalledTimes(1);
+    if (committed) expect((await service().get(row.id))?.metadata).toHaveProperty("archiveProposalError");
+  });
+
+  it("learns another claim written during delivery", async () => {
+    const first = await create(), second = await create(); await claim(first.id, new Date(Date.now() - 120_000));
+    wake.mockImplementationOnce(async () => { await claim(second.id); });
+    await start(); await settled(first.id); await settled(second.id); await quiet();
+    expect(await comments()).toHaveLength(2); expect(wake).toHaveBeenCalledTimes(2);
+  });
+
+  it("finishes the admitted recovery and leaves later claims for restart during shutdown", async () => {
+    const first = await create(), second = await create();
+    await claim(first.id, new Date(Date.now() - 180_000)); await claim(second.id, new Date(Date.now() - 120_000));
+    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); expect(await comments()).toHaveLength(1);
+    expect((await service().get(second.id))?.executionStatus).toBe("running");
+    await start(); await quiet(); expect(await comments()).toHaveLength(2); expect(wake).toHaveBeenCalledTimes(2);
+  });
+});

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

@@ -340,6 +340,8 @@ vi.mock("../services/index.js", () => ({
   })),
   decisionService: vi.fn(() => ({
     sweepExpired: vi.fn(async () => ({ expired: 0 })),
+    recoverEffects: vi.fn(async () => ({ resumed: 0 })),
+    nextEffectRecoveryAt: vi.fn(async () => null),
     sweepTtlExpiry: vi.fn(async () => ({ expired: 0 })),
     nextExpiryAt: vi.fn(async () => null),
     sweepContinuations: vi.fn(async () => ({ retry: false })),
@@ -862,8 +864,10 @@ describe("startServer feedback export wiring", () => {
       const decisions = vi.mocked(decisionService).mock.results.at(-1)!.value;
       expect(decisions.sweepContinuations).toHaveBeenCalledTimes(1);
       expect(decisions.sweepTtlExpiry).toHaveBeenCalledTimes(1);
+      expect(decisions.recoverEffects).toHaveBeenCalledTimes(1);
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionEffectRecovery)).toBe(true);
       expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionExpiry)).toBe(true);
-      expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false, includeExpiry: false });
+      expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false, includeExpiry: false, includeEffects: false });
     } finally {
       setIntervalSpy.mockRestore();
     }
@@ -902,6 +906,7 @@ describe("startServer feedback export wiring", () => {
       expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled();
       expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(false);
       expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionExpiry)).toBe(false);
+      expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionEffectRecovery)).toBe(false);
     } finally {
       setIntervalSpy.mockRestore();
     }

server/src/index.ts

@@ -1,3 +1,4 @@
+import { registerDecisionEffectRecoveryWork } from "./services/decision-effect-recovery-work.js";
 import { registerDecisionExpiryWork } from "./services/decision-expiry-work.js";
 import { registerDecisionContinuationWork } from "./services/decision-continuation-work.js";
 import { registerDecisionArchiveDeliveryWork } from "./services/decision-archive-delivery-work.js";
@@ -1806,7 +1807,9 @@ async function startServerWithDatabaseTeardown(
         () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
       await step("decision_expiry_worker", () => registerDecisionExpiryWork(deliveryWork, decisionExecutor,
         () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
-      await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false }));
+      await step("decision_effect_recovery_worker", () => registerDecisionEffectRecoveryWork(deliveryWork, decisionExecutor,
+        () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
+      await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: 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 },
@@ -1827,7 +1830,7 @@ async function startServerWithDatabaseTeardown(
       trackHeartbeatSchedulerWork((async () => {
         if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
         scheduleRoutineStatusReconcile();
-        trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false }).catch((err: unknown) => {
+        trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false, includeExpiry: false, includeEffects: false }).catch((err: unknown) => {
           logger.error({ err }, "decision expiry sweep failed");
         }));
         trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => {

server/src/services/decision-effect-recovery-work.ts

@@ -0,0 +1,16 @@
+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 registerDecisionEffectRecoveryWork(
+  coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
+  service: Pick<ReturnType<typeof decisionService>, "recoverEffects" | "nextEffectRecoveryAt">,
+  canRun: () => boolean,
+) {
+  return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionEffectRecovery, {
+    retryMs: 30_000, canRun,
+    run: signal => service.recoverEffects(new Date(), undefined, signal),
+    next: service.nextEffectRecoveryAt,
+  });
+}

server/src/services/decision-effect-recovery.ts

@@ -0,0 +1,45 @@
+import { and, asc, lte, sql } from "drizzle-orm";
+import { decisions, type Db } from "@paperclipai/db";
+
+const running = sql`${decisions.status} = 'decided' AND ${decisions.executionStatus} = 'running'`;
+
+export function decisionSweepBatchSize() {
+  const value = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100);
+  return Number.isFinite(value) ? Math.max(1, Math.trunc(value)) : 100;
+}
+
+function recoveryGraceMs() {
+  const value = Number(process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS ?? 60_000);
+  return Number.isFinite(value) && value >= 0 ? value : 60_000;
+}
+
+export function decisionEffectRecoveryService(db: Db, resume: (id: string) => Promise<boolean>) {
+  let cursor: { updatedAt: Date; id: string } | undefined;
+
+  async function sweep(now = new Date(), limit = decisionSweepBatchSize(), signal?: AbortSignal) {
+    let resumed = 0;
+    if (signal?.aborted) return { resumed };
+    const rows = await db.select({ id: decisions.id, updatedAt: decisions.updatedAt }).from(decisions)
+      .where(and(running, lte(decisions.updatedAt, new Date(now.getTime() - recoveryGraceMs())), cursor
+        ? sql`(${decisions.updatedAt}, ${decisions.id}) > (${cursor.updatedAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` : undefined))
+      .orderBy(asc(decisions.updatedAt), asc(decisions.id)).limit(limit);
+    const errors: unknown[] = [];
+    for (const row of rows) {
+      if (signal?.aborted) return { resumed };
+      try { if (await resume(row.id)) resumed++; } catch (error) { errors.push(error); }
+      cursor = row;
+    }
+    if (rows.length < limit) cursor = undefined;
+    if (errors.length === 1) throw errors[0];
+    if (errors.length) throw new AggregateError(errors, "Decision effect recovery failed");
+    return { resumed };
+  }
+
+  async function next(): Promise<number | null> {
+    const [row] = await db.select({ at: decisions.updatedAt }).from(decisions).where(running)
+      .orderBy(asc(decisions.updatedAt), asc(decisions.id)).limit(1);
+    return row ? row.at.getTime() + recoveryGraceMs() : null;
+  }
+
+  return { sweep, next };
+}

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

@@ -4,6 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
 export async function notifyDecisionChange(transaction: object) {
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation);
   await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionExpiry);
+  await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionEffectRecovery);
 }
 
 export function writeDecisionChange<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {

server/src/services/decisions.ts

@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { and, asc, count, desc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm";
+import { and, asc, count, desc, eq, gt, gte, inArray, sql } from "drizzle-orm";
 import type { Db } 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";
@@ -15,6 +15,7 @@ import {
 import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js";
 
 import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js";
+import { decisionEffectRecoveryService, decisionSweepBatchSize } from "./decision-effect-recovery.js";
 import { decisionExpiryService } from "./decision-expiry.js";
 import { notifyDecisionChange, writeDecisionChange } from "./decision-work-notifications.js";
 
@@ -126,6 +127,7 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) {
 export function decisionService(db: Db, options: DecisionServiceOptions) {
   const continuations = decisionContinuationService(db, options.wakeOriginAgent);
   const expiry = decisionExpiryService(db, continuations.deliver);
+  const effectRecovery = decisionEffectRecoveryService(db, recoverDecisionEffects);
   const scope = databaseWorkScopeKey(db);
   let flights = effectFlights.get(scope);
   if (!flights) effectFlights.set(scope, flights = new Map());
@@ -716,27 +718,18 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
       return { ...bundle, decisions: created }; });
   }
 
-  async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean } = {}) {
-    const configuredBatchSize = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100);
-    const batchSize = Number.isFinite(configuredBatchSize)
-      ? Math.max(1, Math.trunc(configuredBatchSize))
-      : 100;
-    const configuredRecoveryGraceMs = Number(process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS ?? 60_000);
-    const recoveryGraceMs = Number.isFinite(configuredRecoveryGraceMs) && configuredRecoveryGraceMs >= 0
-      ? configuredRecoveryGraceMs
-      : 60_000;
-    const runningRows = await db.select().from(decisions)
-      .where(and(eq(decisions.status, "decided"), eq(decisions.executionStatus, "running"),
-        lte(decisions.updatedAt, new Date(now.getTime() - recoveryGraceMs))))
-      .orderBy(asc(decisions.updatedAt)).limit(batchSize);
-    let resumed = 0;
-    for (const decision of runningRows) {
-      if (!decision.decidedByUserId) continue;
-      const recovered = await runEffects(decision, await recoveryActor(decision.companyId, decision.decidedByUserId));
-      if (recovered.executionStatus === "running") continue;
-      resumed += 1;
-      await continuations.deliver(recovered.id);
-    }
+  async function recoverDecisionEffects(id: string) {
+    const current = await get(id);
+    if (!current || current.status !== "decided" || current.executionStatus !== "running" || !current.decidedByUserId) return false;
+    const recovered = await runEffects(current, await recoveryActor(current.companyId, current.decidedByUserId));
+    if (recovered.executionStatus === "running") return false;
+    await continuations.deliver(recovered.id);
+    return true;
+  }
+
+  async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean; includeExpiry?: boolean; includeEffects?: boolean } = {}) {
+    const batchSize = decisionSweepBatchSize();
+    const { resumed } = options.includeEffects === false ? { resumed: 0 } : await effectRecovery.sweep(now, batchSize);
     if (options.includeContinuations !== false) await continuations.sweep(batchSize);
     const ttlRows = options.includeExpiry === false ? [] : await expiry.due(now, batchSize);
     const remaining = batchSize - ttlRows.length;
@@ -768,5 +761,6 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
   }
 
   return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep,
-    sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next };
+    sweepTtlExpiry: expiry.sweep, nextExpiryAt: expiry.next,
+    recoverEffects: effectRecovery.sweep, nextEffectRecoveryAt: effectRecovery.next };
 }

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

@@ -8,6 +8,7 @@ export const DELIVERY_QUEUES = {
   executionStatus: "execution-status",
   executionDisposition: "execution-disposition",
   environmentCleanup: "environment-cleanup",
+  decisionEffectRecovery: "decision-effect-recovery",
   decisionExpiry: "decision-expiry",
   decisionContinuation: "decision-continuation",
   decisionArchiveDelivery: "decision-archive-delivery",