fix(server): serialize decision outcomes and bound continuation shutdown

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

doc/idle-decision-continuations.md

@@ -27,8 +27,9 @@ 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.
+Each worker pass attempts at most 100 candidates. A saved in-memory cursor
+advances past failed candidates between passes and wraps after the last page.
+Remaining durable work retains another pass. This avoids starving later rows behind a failing first page.
 A partial ID index contains ready pending continuations only. The predicate uses
 fixed SQL literals so a prepared generic plan can still use that index.
 
@@ -45,7 +46,9 @@ 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,
+The promise is removed on success or failure. Effect execution and its final
+state write also share one local flight per decision. This prevents concurrent
+archive callers from turning a successful archive into a stale-manifest failure. This is process-local coalescing,
 not a new durable claim or an exactly-once guarantee.
 
 Effect completion requires the row to remain running. A competing completion
@@ -66,7 +69,9 @@ 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.
+waits for the current admitted delivery and stops before starting another.
+The shared deadline adapter forwards its existing AbortSignal to task execution;
+existing callbacks that do not consume the signal keep their behavior. A future process-local retry is not a Cloud alarm.
 
 Enabled-heartbeat startup registers this worker before the combined decision
 sweep. The main tick skips that sweep's independent continuation scan. Its public
@@ -81,7 +86,7 @@ host/database suspension or measured fleet savings.
 
 ## Verification
 
-The selected run passes 92 tests: 49 real-PostgreSQL decision cases, 39 startup
+The selected run passes 93 tests: 50 real-PostgreSQL decision cases, 39 startup
 cases, and four database work-signal unit cases. Fifteen new continuation cases
 cover real decide/cancel/dismiss/expiry producers, running-to-terminal completion,
 foreground contention, dedicated child pools, stale caller re-read, restart,
@@ -91,7 +96,7 @@ 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.
+passes all 49 decision tests before the additional archive-race regression.
 
 Migration 0382 applies and replays without error. A rolled-back synthetic
 40,000-row history with one pending continuation uses the partial index under
@@ -104,3 +109,18 @@ message at the top level. The assertion now checks the original cause. No
 production retry, hold, or timeout was weakened. A recursive unit-fixture type
 initially prevented migration generation; an explicit fixture type fixed it.
 Full CI and fresh review remain required before PR readiness.
+
+
+Review found two real regressions. Before repair, a controlled concurrent archive
+reported failed after archiving successfully, and shutdown sent a second wake
+after stop was requested. Both regressions failed before the fix. Shared local
+effect ownership fixes archive completion; bounded passes and abort propagation
+fix backlog shutdown. The 93-case selection passes after repair. The paging test
+also proves a failed first candidate cannot hide the 101st candidate across
+bounded passes. A result-type alias initially made the known decision ID optional;
+the alias now matches the existing outcome contract.
+
+The existing tool-review and archive-delivery consumers also pass 124 real-DB
+tests after AbortSignal propagation. Together with the decision/startup/signal
+selection, 217 selected tests pass; the same nine older embedded-only cases
+remain locally skipped. Workspace typecheck passes after the outcome type fix.

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 | Pending |
+| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations: [#17478](https://github.com/paperclipai/paperclip/pull/17478), review repairs under verification; expiry branch in progress; 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 |
@@ -392,3 +392,21 @@ Parent #17475 at `692e3ee02c` now has passing checks and Greptile 5/5; the extra
 Paperclip Review check was skipped. Fresh master is `fe85fd7ac4`, still at
 migration 0377. The stack remains open and unmerged. The full inventory above
 remains the completion scope.
+
+
+Slice 09 is [#17478](https://github.com/paperclipai/paperclip/pull/17478), initially
+`66bbef521e`. Review found a concurrent archive result race and unbounded backlog
+shutdown. Both failed real-boundary regressions before repair. The local repair
+shares effect execution through completion per DB owner/decision, bounds each
+continuation pass to 100 with a fair cursor, and forwards cancellation through
+the shared deadline adapter. The repaired selection passes 217 tests, with nine
+older embedded-only signal tests skipped locally. Typecheck passes. Fresh build,
+CI, and review remain required; the initial head has no other CI failure.
+
+The separate TTL branch `codex/sleep-10-decision-expiry` contains implementation
+commit `18c11e3110`. It passed 101 selected tests, workspace typecheck and build.
+Its fault fixture initially cleared its error collector before a previous
+admitted failure finished; the assertion now verifies all errors after settled
+recovery. No hold or retry was weakened. This branch must receive the parent
+review repairs and adapt its own paged expiry worker to cancellation before PR
+publication. No tenth PR or readiness is claimed.

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

@@ -24,6 +24,7 @@ import {
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { attentionService } from "../services/attention.js";
 import { decisionService } from "../services/decisions.js";
+import * as retentionModule from "../services/decision-retention.js";
 import { hashAttentionArchiveManifest } from "../services/decision-retention.js";
 import type { AttentionArchiveManifestEntry, AttentionArchiveTargetSnapshot } from "@paperclipai/shared";
 
@@ -206,6 +207,45 @@ describePg("decisionService", () => {
     expect(states.find((row) => row.sourceId === unreviewedIssueId)?.archivedAt).toBeNull();
   });
 
+  it("serializes concurrent archive completion before reporting its outcome", async () => {
+    await db.update(issues).set({ status: "in_review", assigneeUserId: decidedByUserId }).where(eq(issues.id, targetIssueId));
+    const activityAt = new Date("2026-04-01T00:00:00.000Z");
+    await db.insert(decisionRetention).values({ companyId, sourceKind: "review", sourceId: targetIssueId, sourceActivityAt: activityAt });
+    const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetIssueId,
+      expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }];
+    const row = await service().create({ companyId, actor: agentActor(), agentId, runId, title: "Archive?", body: "Exact set",
+      continuationPolicy: "wake_origin_agent", options: [{ id: "archive", label: "Archive", effects: [] }],
+      metadata: { kind: "attention_archive_proposal", manifestHash: hashAttentionArchiveManifest(manifest) },
+      additionalTargetSnapshots: { [`attention:review:${targetIssueId}`]: { status: "attention", assigneeAgentId: null,
+        assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } });
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    const original = retentionModule.decisionRetentionService;
+    let archived = false, calls = 0;
+    const factory = vi.spyOn(retentionModule, "decisionRetentionService").mockImplementation((source, options) => {
+      const actual = original(source, options);
+      return { ...actual, archiveReviewedManifest: async input => {
+        calls++;
+        const result = await actual.archiveReviewedManifest(input);
+        archived = true; await gate; return result;
+      } };
+    });
+    const decide = () => service().decide({ id: row.id, optionId: "archive", decidedByUserId, userActor: boardActor() });
+    const first = decide();
+    let second: ReturnType<typeof decide> | undefined;
+    try {
+      await vi.waitFor(() => expect(archived).toBe(true));
+      second = decide();
+      await new Promise(resolve => setTimeout(resolve, 100));
+      release();
+      const results = await Promise.all([first, second]);
+      expect(results.map(result => result.executionStatus)).toEqual(["succeeded", "succeeded"]);
+      expect(calls).toBe(1); expect(wakes).toHaveLength(1);
+      const [state] = await db.select().from(decisionRetention);
+      expect(state.archiveVersion).toBe(1); expect(state.archivedAt).not.toBeNull();
+    } finally { release(); await Promise.allSettled([first, ...(second ? [second] : [])]); factory.mockRestore(); }
+  });
+
   it("skips strict stale targets and fails closed on intersection denial", async () => {
     const stale = await createCommentDecision("strict");
     await db.update(issues).set({ updatedAt: new Date(Date.now() + 1_000) }).where(eq(issues.id, targetIssueId));
@@ -843,7 +883,8 @@ describePg("decisionService", () => {
     });
 
     it("awaits an admitted wake before completing shutdown", async () => {
-      const row = await createCommentDecision(); await markReady(row.id);
+      const row = await createCommentDecision(), other = await createCommentDecision();
+      await markReady(row.id); await markReady(other.id);
       let release!: () => void;
       const gate = new Promise<void>(resolve => { release = resolve; });
       wake.mockImplementationOnce(async () => gate);
@@ -857,7 +898,9 @@ describePg("decisionService", () => {
         expect(stopped).toBe(false);
       } finally { release(); }
       await starting; await stopping!; coordinator = undefined;
-      await delivered(row.id); expect(stopped).toBe(true);
+      expect(wake).toHaveBeenCalledTimes(1);
+      const sentId = wake.mock.calls[0][0].decisionId as string;
+      await delivered(sentId); await pending(sentId === row.id ? other.id : row.id); expect(stopped).toBe(true);
     });
 
     it("pages past a failed first candidate and preserves its retry", async () => {
@@ -867,10 +910,13 @@ describePg("decisionService", () => {
       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();
+      const recovery = workerService();
+      await expect(recovery.sweepContinuations()).rejects.toThrow("first unavailable");
+      expect(wake).toHaveBeenCalledTimes(100); await pending(ids.at(-1)!);
+      expect(await recovery.sweepContinuations()).toEqual({ retry: true });
       expect(wake).toHaveBeenCalledTimes(101); await delivered(ids.at(-1)!); await pending(ids[0]);
-      expect(errors).toHaveLength(1); errors.length = 0; wake.mockImplementation(async () => {});
-      await quiet(); await delivered(ids[0]); expect(wake).toHaveBeenCalledTimes(102);
+      wake.mockImplementation(async () => {});
+      await start(); await quiet(); await delivered(ids[0]); expect(wake).toHaveBeenCalledTimes(102);
     });
   });
 

server/src/services/deadline-work.ts

@@ -5,7 +5,7 @@ export function registerDeadlineWork(
   coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
   queue: DeliveryQueue,
   task: {
-    run: () => Promise<unknown>;
+    run: (signal: AbortSignal) => Promise<unknown>;
     next: () => Promise<number | null>;
     canRun: () => boolean;
     retryMs: number;
@@ -18,12 +18,13 @@ export function registerDeadlineWork(
     retryMs: task.retryMs,
     pausePassiveMaintenance: true,
     onSettled: () => { requested++; },
-    run: async () => {
+    run: async signal => {
       if (!task.canRun()) { requested++; return; }
       if (Date.now() < earliest) { next = earliest; return; }
       const generation = ++requested;
       earliest = Date.now() + task.retryMs;
-      await task.run();
+      await task.run(signal);
+      if (signal.aborted) return;
       const deadline = await task.next();
       next = deadline === null ? null : Math.max(earliest, deadline);
       reconciled = generation;

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

@@ -11,8 +11,8 @@ export function registerDecisionContinuationWork(
   let next: number | null = null;
   return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionContinuation, {
     retryMs: 30_000, canRun,
-    run: async () => {
-      const result = await service.sweepContinuations();
+    run: async signal => {
+      const result = await service.sweepContinuations(100, signal);
       next = result.retry ? Date.now() + 30_000 : null;
     },
     next: async () => next,

server/src/services/decision-continuations.ts

@@ -17,6 +17,7 @@ export function decisionContinuationService(db: Db, wake: DecisionContinuationWa
   let pending = flights.get(scope);
   if (!pending) flights.set(scope, pending = new Map());
   const inFlight = pending;
+  let recoveryCursor: string | undefined;
 
   function deliver(id: string): Promise<void> {
     const existing = inFlight.get(id);
@@ -37,24 +38,26 @@ export function decisionContinuationService(db: Db, wake: DecisionContinuationWa
     return attempt;
   }
 
-  async function sweep(maxRows = Infinity) {
-    let cursor: string | undefined;
+  async function sweep(maxRows = 100, signal?: AbortSignal) {
     let attempted = 0;
     const errors: unknown[] = [];
     while (attempted < maxRows) {
+      if (signal?.aborted) return { retry: true };
       const limit = Math.min(100, maxRows - attempted);
       const rows = await db.select({ id: decisions.id }).from(decisions)
-        .where(and(ready, cursor ? gt(decisions.id, cursor) : undefined))
+        .where(and(ready, recoveryCursor ? gt(decisions.id, recoveryCursor) : undefined))
         .orderBy(asc(decisions.id)).limit(limit);
       for (const row of rows) {
+        if (signal?.aborted) return { retry: true };
         try { await deliver(row.id); } catch (error) { errors.push(error); }
+        recoveryCursor = row.id;
       }
       attempted += rows.length;
-      if (rows.length < limit) break;
-      cursor = rows.at(-1)!.id;
+      if (rows.length < limit) { recoveryCursor = undefined; break; }
     }
     if (errors.length === 1) throw errors[0];
     if (errors.length) throw new AggregateError(errors, "Decision continuation delivery failed");
+    if (signal?.aborted) return { retry: true };
     const remaining = await db.select({ id: decisions.id }).from(decisions).where(ready).limit(1);
     return { retry: remaining.length > 0 };
   }

server/src/services/decisions.ts

@@ -1,7 +1,7 @@
 import { randomUUID } from "node:crypto";
 import { and, asc, count, desc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
-import { companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db";
+import { databaseWorkScopeKey, companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db";
 import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared";
 import type { AttentionArchiveManifestEntry, DecisionEffect, DecisionInput, DecisionOption, DecisionStatsCounts, DecisionStatsResponse } from "@paperclipai/shared";
 import { conflict, forbidden, notFound, tooManyRequests, unprocessable } from "../errors.js";
@@ -22,6 +22,8 @@ type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId
 
 export type DecisionServiceOptions = { wakeOriginAgent: DecisionContinuationWake };
 const DAY = 86_400_000;
+type DecisionOutcome = typeof decisions.$inferSelect & { executions: Array<typeof decisionEffectExecutions.$inferSelect> };
+const effectFlights = new WeakMap<object, Map<string, Promise<DecisionOutcome>>>();
 
 function targetIds(options: DecisionOption[]) {
   const result = new Set<string>();
@@ -122,6 +124,10 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) {
 
 export function decisionService(db: Db, options: DecisionServiceOptions) {
   const continuations = decisionContinuationService(db, options.wakeOriginAgent);
+  const scope = databaseWorkScopeKey(db);
+  let flights = effectFlights.get(scope);
+  if (!flights) effectFlights.set(scope, flights = new Map());
+  const executing = flights;
   const authz = authorizationService(db);
   let targetSweepCursor: string | null = null;
   type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null;
@@ -548,7 +554,19 @@ export function decisionService(db: Db, options: DecisionServiceOptions) {
     }
   }
 
-  async function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
+  function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor): Promise<DecisionOutcome> {
+    const existing = executing.get(decision.id);
+    if (existing) return existing;
+    const attempt = (async () => {
+      const current = await get(decision.id);
+      if (!current) throw notFound("Decision not found");
+      return current.executionStatus === "running" ? executeDecisionEffects(current, userActor) : outcome(current.id);
+    })().finally(() => { executing.delete(decision.id); });
+    executing.set(decision.id, attempt);
+    return attempt;
+  }
+
+  async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) {
     const option = decision.options.find((item) => item.id === decision.chosenOptionId);
     if (!option || !decision.decidedByUserId) throw unprocessable("Stored decision outcome is invalid");
     const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null);