diff --git a/doc/idle-decision-continuations.md b/doc/idle-decision-continuations.md index 8146e4ff653..54c028f0bb9 100644 --- a/doc/idle-decision-continuations.md +++ b/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. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index f7733fcd0f7..673b1f1d01b 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/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. diff --git a/server/src/__tests__/decisions-service.test.ts b/server/src/__tests__/decisions-service.test.ts index f235469e19c..aaa3de3e3ba 100644 --- a/server/src/__tests__/decisions-service.test.ts +++ b/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(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 | 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(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); }); }); diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index 2e39dc4c4c3..bb3c74a4f0b 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -5,7 +5,7 @@ export function registerDeadlineWork( coordinator: ReturnType, queue: DeliveryQueue, task: { - run: () => Promise; + run: (signal: AbortSignal) => Promise; next: () => Promise; 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; diff --git a/server/src/services/decision-continuation-work.ts b/server/src/services/decision-continuation-work.ts index 78dfbb6eef2..03595fd0d0c 100644 --- a/server/src/services/decision-continuation-work.ts +++ b/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, diff --git a/server/src/services/decision-continuations.ts b/server/src/services/decision-continuations.ts index d703273560f..79c70b43969 100644 --- a/server/src/services/decision-continuations.ts +++ b/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 { 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 }; } diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index 280ccee9e4c..177256ed9fb 100644 --- a/server/src/services/decisions.ts +++ b/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 }; +const effectFlights = new WeakMap>>(); function targetIds(options: DecisionOption[]) { const result = new Set(); @@ -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 { + 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);