diff --git a/doc/idle-decision-continuations.md b/doc/idle-decision-continuations.md new file mode 100644 index 00000000000..c801ccf2e20 --- /dev/null +++ b/doc/idle-decision-continuations.md @@ -0,0 +1,193 @@ +# Decision continuation delivery while idle + +A terminal decision can owe an agent wake. Its existing row stores that debt: +`continuationPolicy = wake_origin_agent`, `metadata.continuationPending = true`, +and either expired/cancelled status or a decided status with succeeded/partial/ +failed execution. Open decisions and running effects are not ready to deliver. +This migration adds no queue table and changes no wake payload. + +## Ownership and producers + +`decisions.ts` owns decision creation, authorization, signatures, effects, and +state changes. Its ten update sites now use a narrow transaction-aware write +helper before the SQL statement. These cover the running claim; ordinary effect +completion; archive-proposal signature, manifest, and execution failures; archive +success; TTL rejection on use; cancel; direct dismissal; and background expiry. +The helper acquires intent in the existing outer transaction or its savepoint. +It does not include effect execution, provider calls, activity publication, or +agent wake callbacks in a new database transaction. + +Creation and bundle creation insert open decisions, so they do not create ready +continuation debt. Acknowledgement only clears debt and belongs to the delivery +owner; it does not signal itself. The production decision services are created +by server startup and routes with the application's DB handle. Independent SQL +writers and additional active processes require a separate notification and +ownership contract. This change does not make wake-producing service methods +safe to call inside an uncommitted caller transaction; only their SQL mutation +boundary has the existing outer-commit guarantee. + +`decision-continuations.ts` owns the readiness predicate, delivery, and recovery. +Each worker pass attempts at most 100 candidates. A saved in-memory cursor +advances past failed candidates between passes and wraps after the last page. +Remaining durable work retains another pass. This avoids starving later rows behind a failing first page. +A partial ID index contains ready pending continuations only. The predicate uses +fixed SQL literals so a prepared generic plan can still use that index. + +The shared DB package exposes an opaque work-scope identity. Raw DB object +identity is insufficient because dedicated and liveness child pools already +share their owner's signals. The key is shared by those children and nested +transactions; independent roots have different keys. No connection credentials, +private transaction state, or decision policy are exposed. Existing signal +consumers and transactions that do not register work keep their behavior. + +## Foreground, recovery, and failure + +Foreground delivery remains immediate. All service instances sharing the DB +owner coalesce a simultaneous wake for the same decision into one promise. +Delivery re-reads durable readiness after claiming that local flight, so a late +foreground call cannot resend a continuation already acknowledged by recovery. +The promise is removed on success or failure. Effect execution and its final +state write also share one local flight per decision. This prevents concurrent +archive callers from turning a successful archive into a stale-manifest failure. This is process-local coalescing, +not a new durable claim or an exactly-once guarantee. + +Effect completion requires the row to remain running. A competing completion +cannot restore pending metadata after another caller has completed and delivered +it. Effects retain their own existing durable claims and authorization checks. +The delivery worker never executes effects. + +A wake failure leaves pending metadata intact. A failed acknowledgement also +leaves debt, so a later attempt can send the wake again. A crash after accepting +the wake but before acknowledgement retains the existing at-least-once contract. +An uncertain producer transaction remains held until the shared work-signal +protocol establishes settlement; one empty scan cannot clear that uncertainty. + +## Lifecycle and scope + +Startup reconstructs pending delivery. Committed changes wake the worker through +the existing coordinator. Failures or remaining ready debt retain a 30-second +retry. Open/running decisions, no decisions, and delivered history have no +recurring continuation scan or deadline. Changes during a pass retain another +request. Drain/standby defer dispatch without consuming that request; shutdown +waits for the current admitted delivery and stops before starting another. +The shared deadline adapter forwards its existing AbortSignal to task execution; +existing callbacks that do not consume the signal keep their behavior. A future process-local retry is not a Cloud alarm. + +Enabled-heartbeat startup registers this worker before the combined decision +sweep. The main tick skips that sweep's independent continuation scan. Its public +API retains combined behavior by default. Expiry or effect recovery can still +deliver their newly produced continuations immediately. Disabled-heartbeat +startup keeps its existing behavior and does not dispatch this new worker. + +Decision expiry, invalidated-target scanning, interrupted effect recovery, and +retention remain separate migrations. Existing Cloud safety blockers are not +relaxed. This establishes worker quiet under the named conditions, not deployed +host/database suspension or measured fleet savings. + +## Verification + +The selected run passes 93 tests: 50 real-PostgreSQL decision cases, 39 startup +cases, and four database work-signal unit cases. Fifteen new continuation cases +cover real decide/cancel/dismiss/expiry producers, running-to-terminal completion, +foreground contention, dedicated child pools, stale caller re-read, restart, +failed acknowledgement, outer rollback/commit, first write during drain, failed +retry through drain, changes during delivery, shutdown, and paging past a failed +candidate. Quiet checks observe no SQL calls across an hour of controlled time. +Nine pre-existing embedded-PostgreSQL signal tests are skipped locally because +the embedded runtime is unavailable; CI must run that supported environment. +Workspace typecheck and build pass. The final literal-predicate adjustment also +passes all 49 decision tests before the additional archive-race regression. + +The original continuation index applied and replayed without error. After parent integration it is generated migration `0388_ambitious_micromax.sql`; the fresh apply and replay also pass. A rolled-back synthetic +40,000-row history with one pending continuation uses the partial index under +`force_generic_plan`, reads two shared buffers, and executes in 0.015 ms on the +local PostgreSQL server. This is query evidence, not a production latency claim. + +The first test selection had 90 passes and one failure: the injected PostgreSQL +acknowledgement error is wrapped by Drizzle, but the assertion expected its +message at the top level. The assertion now checks the original cause. No +production retry, hold, or timeout was weakened. A recursive unit-fixture type +initially prevented migration generation; an explicit fixture type fixed it. +Full CI and fresh review remain required before PR readiness. + + +Review found two real regressions. Before repair, a controlled concurrent archive +reported failed after archiving successfully, and shutdown sent a second wake +after stop was requested. Both regressions failed before the fix. Shared local +effect ownership fixes archive completion; bounded passes and abort propagation +fix backlog shutdown. The 93-case selection passes after repair. The paging test +also proves a failed first candidate cannot hide the 101st candidate across +bounded passes. A result-type alias initially made the known decision ID optional; +the alias now matches the existing outcome contract. + +The existing tool-review and archive-delivery consumers also pass 124 real-DB +tests after AbortSignal propagation. Together with the decision/startup/signal +selection, 217 selected tests pass; the same nine older embedded-only cases +remain locally skipped. Workspace typecheck passes after the outcome type fix. + + +## Explicit archive-failure delivery + +The three archive-proposal failure paths now explicitly set +`continuationPending: true` for `wake_origin_agent`, rather than assuming the +pre-read running row already carries that metadata. The existing running-state +fence remains, so stale settlement cannot re-arm an acknowledged result. +Invalid signatures, manifest mismatches and archive execution failures each +failed a real-PostgreSQL recovery regression before repair. Each now saves the +failed result, wakes once, acknowledges delivery and remains quiet on replay. + +Normal parent integration preserves the projection interaction-link index and +archive pending-order index. The unmerged continuation index was regenerated +as 0388 from the parent's generated metadata; every parent journal entry is +unchanged, five snapshots remain, the SQL uses the existing top-level concurrent +executor, and snapshot/schema drift passes. No generated snapshot was hand-edited. + +The final affected selection passes 111 cases: 53 decisions, 15 retention/archive, +42 startup and one scoped-deadline case, with all four suites exercised. The +first combined command used the wrong retention DB variable and skipped those +15 cases; the corrected external-PostgreSQL retention run passes all 15. Server +typecheck and build pass. The historical nine embedded-only work-signal skips +above are not reclassified as executed by this selection. Fresh CI and reviews +remain required. + + +## Bounded notification-driven recovery + +Security review identified that sequential decision writes could each start an +instance-wide continuation scan. A real-producer regression reproduced six +sweeps for six sequential cancellations. The worker now uses the existing +`notificationCoalesceMs: 30_000` option. The first wake after a quiet window still +reconciles promptly, and subsequent notifications in that window share its end +deadline. Foreground continuation delivery is unchanged and immediate. Recovery +can wait up to thirty seconds, matching the prior maintenance cadence. This +bounds global scan frequency; it does not make queries company-scoped or add a +cross-process notification contract. + +The regression verifies one scan during repeated real create/cancel writes, +retention and eventual delivery of a later pending row, and no recurring SQL +after settlement. Its first green attempt wrongly expected zero active work +while the coalesced request was pending. The existing intent hold correctly +remains until reconciliation; the test now checks that hold's actual terminal +boundary through the existing quiet assertion. No production hold was removed. +All 293 integrated cases across decisions, archive, gateway/projection, startup +and scoped-deadline suites pass without skips; server types/build pass. Fresh +security review, Greptile, Ponytail and exact-head CI remain merge gates. + + +The archive contention regression no longer waits a fixed 100 ms. A dedicated +single-connection contender signals its decision read; two FIFO SQL round trips +then establish that it reached shared execution while the first archive remains +held. The test requires one contender decision read and one archive execution. +Temporarily removing the effect-flight guard made this test fail (two reads); +restoring the production guard passes. The mutation was restored before commit. +This verifies actual cross-pool contention without a timing delay or a new +production-only test hook. + + +Archive #17475 is merged. Normal reconciliation with its exact squash +`5785090ff1` preserves all 31 independent incoming master files, including +Runner controller-authority recovery. All 97 decision/startup/deadline cases and +server types/build pass after that integration. The Runner vendor rebuilt from +its changed inputs. The preceding full five-suite selection passed 293 cases; +the continuation owner's production behavior is unchanged by reconciliation. +Fresh exact-head reviews and CI remain required on the master-target candidate. 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 2345ab9bb90..6b1b248598f 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -15,10 +15,12 @@ Do not mark an item done from a timer removal or a green check alone. | 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | [#17443](https://github.com/paperclipai/paperclip/pull/17443), merged after CI and passing reviews | | 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | -| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; notification delivery implemented in slice 08, PR pending | -| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | +| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Retention pending; archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged after exact-head 5/5 reviews and passing current CI | +| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478), expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479), and effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) await fresh current-head gates; target invalidation review repair remains in progress | + | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | -| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), merged after current-head CI and both 5/5 reviews; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), merged after current-head CI and both 5/5 reviews; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), interaction-link index verified and awaiting fresh CI/review | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Slices 02–07 merged: [secret proposals](https://github.com/paperclipai/paperclip/pull/17446), [credentials](https://github.com/paperclipai/paperclip/pull/17450), [review expiry](https://github.com/paperclipai/paperclip/pull/17453), [approved recovery](https://github.com/paperclipai/paperclip/pull/17454), [interrupted claims](https://github.com/paperclipai/paperclip/pull/17458), [projection](https://github.com/paperclipai/paperclip/pull/17468) | + | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | @@ -359,3 +361,131 @@ counter regression and corrected disabled-mode expectation are recorded in the behavior document. Full current-head CI and both reviews remain required. Retention eligibility/scanning, decision expiry, effect recovery, and decision continuation delivery remain independent unfinished reasons in the full ledger. + + +Slice 08 is [#17475](https://github.com/paperclipai/paperclip/pull/17475), head +`692e3ee02c`, based on #17468. The PR is attached and Greptile review was +requested. Fresh CI is active with no failures in the latest snapshot. +The first seven PRs are mergeable and have all checks passing at their recorded +heads. They remain open; no merge was requested for this stack. + +## Next owner audit: decision continuations + +The next branch is `codex/sleep-09-decision-continuations`, based on #17475. +No ninth implementation or readiness is claimed. `decisions.ts:sweepExpired` +currently mixes four distinct reasons: interrupted effect recovery, pending +continuation delivery, TTL expiry, and invalidated strict targets. Preserve +separate PRs for those duties. A deadline alone cannot replace target-change +notifications; target deletion and completion need an audited producer contract. + +For continuation delivery, the existing decision row already stores the durable +obligation: `continuationPolicy = wake_origin_agent`, +`metadata.continuationPending = true`, and a terminal outcome. Decided rows +are ready only after execution reaches succeeded, partial, or failed. Open +decisions and still-running effects must not create a delivery retry. + +All current decision-row writers are in `decisions.ts`. Production service +instances are created in server startup and decision routes with the server DB +handle. Audit creation/bundles, decision claims, regular and archive-proposal +effect settlement/failure, TTL expiry on use, background expiry, cancellation, +dismissal, and continuation acknowledgement. Reuse a narrow pre-write helper +inside existing transaction ownership; do not wrap provider/effect execution +or agent wake calls in a database transaction. The archive-proposal error +branches inherit the claim's pending metadata and must remain eligible after +terminal failure. + +The current foreground paths call `deliverContinuation` after saving the result. +It sends the wake then clears pending metadata. Simply adding an eager worker +would make foreground/worker overlap common. Preserve prompt foreground delivery +and the existing at-least-once crash contract, but coalesce simultaneous attempts +for the same decision and re-read durable readiness before sending. Scope any +in-memory coalescing to the actual database/instance owner; do not let unrelated +instances or differing transaction ownership share an accidental global key. +The DB signal owner also supports child pools, so a proposed coalescing mechanism +must explicitly reconcile that boundary rather than assuming object identity +is always sufficient. No new table is needed merely for this migration. + +After commit-aware producer coverage, add a paged continuation-only recovery +entry point and register it with the existing coordinator. Remove only its +independent pending scan from the periodic combined sweep. Keep normal +continuations produced by expiry/effect completion, and preserve the public +combined recovery entry point for current callers/tests. A failed wake or +acknowledgement retains its durable pending state and a retry. Matching delivered +history becomes quiet. + +Tests should exercise actual decide/cancel/dismiss/expiry producers, completion +of effects after a running claim, concurrent foreground and worker delivery, +wake/acknowledgement failures, restart, outer commit/rollback, first write and +failure during drain, pagination past failing rows, and delivered-history quiet. +The existing decisions suite has an embedded-only setup; use the repository's +validated external test database helper to exercise it on the disposable local +PostgreSQL server, without silently skipping the suite when embedded setup is +unavailable. + +Slice 08 review update: Greptile is 5/5 on `692e3ee02c` with no actionable +findings. The latest CI snapshot has 10 successful checks and 44 in progress, +with no failures. This is a live CI wait, not a readiness or completion claim. + + +## Slice 09 implementation: decision continuations + +The continuation owner and worker are implemented on `codex/sleep-09-decision-continuations`. +See [decision continuations](../idle-decision-continuations.md) for producer, +transaction, concurrency, failure, and test evidence. The existing row is durable +truth; the only DB API addition is an opaque shared work-scope identity for local +coalescing across child pools. No new queue table or external wake exemption is +added. Terminal effect settlement is fenced to running state so a competing +caller cannot re-arm an acknowledged continuation. A concurrent partial index +bounds pending selection independently of retained history. + +The selected run passes 92 tests; nine older embedded-Postgres cases are locally +skipped, with the new lifecycle tests running against real external PostgreSQL. +Workspace typecheck/build pass. Migration apply/replay and a 40,000-row prepared +query probe pass. Initial fixture type/error-shape failures are recorded in the +behavior document. PR creation and full CI/review are next. + +Parent #17475 at `692e3ee02c` now has passing checks and Greptile 5/5; the extra +Paperclip Review check was skipped. Fresh master is `fe85fd7ac4`, still at +migration 0377. The stack remains open and unmerged. The full inventory above +remains the completion scope. + + +Slice 09 is [#17478](https://github.com/paperclipai/paperclip/pull/17478), initially +`66bbef521e`. Review found a concurrent archive result race and unbounded backlog +shutdown. Both failed real-boundary regressions before repair. The local repair +shares effect execution through completion per DB owner/decision, bounds each +continuation pass to 100 with a fair cursor, and forwards cancellation through +the shared deadline adapter. The repaired selection passes 217 tests, with nine +older embedded-only signal tests skipped locally. Typecheck passes. Fresh build, +CI, and review remain required; the initial head has no other CI failure. + +The separate TTL branch `codex/sleep-10-decision-expiry` contains implementation +commit `18c11e3110`. It passed 101 selected tests, workspace typecheck and build. +Its fault fixture initially cleared its error collector before a previous +admitted failure finished; the assertion now verifies all errors after settled +recovery. No hold or retry was weakened. This branch must receive the parent +review repairs and adapt its own paged expiry worker to cancellation before PR +publication. No tenth PR or readiness is claimed. + + +Slice 09 review follow-up: every failed archive-proposal settlement explicitly +creates its origin-agent continuation obligation. Three recovered-row regressions +failed before repair and pass with exactly one wake and persisted acknowledgement. +The current parent-integrated selection passes 111 decision/archive/startup/deadline +cases, server types/build, fresh migration apply/replay and generated snapshot +drift. Own unmerged index is regenerated as `0388_ambitious_micromax.sql`; parent +386/387 migrations and journal entries are preserved. The initial skipped retention +command and corrected 15-case run are recorded in the behavior document. + +Slice 09 security follow-up: sequential real decision writes reproduced six +global continuation scans. Existing thirty-second notification coalescing now +bounds those scans while preserving immediate foreground wakes and durable +pending recovery. All 293 integrated decision/archive/gateway/startup/deadline +cases and server types/build pass. The incorrect expectation of zero active +work during the coalescing window is recorded in the behavior document; the +intent hold remains until reconciliation. Fresh security/CI/reviews required. + +Eight slices are merged after archive delivery #17475. Slice 09 now reconciles +with its exact squash `5785090ff1`; all 31 independent master files are preserved. +The 97 decision/startup/deadline cases and server types/build pass after the +Runner vendor rebuild. Fresh master-target CI and reviews remain required. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 1d98d929d7a..443005ca277 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -506,6 +506,8 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0385_open_madame_web.sql", "0386_steady_epoch.sql", "0387_boring_professor_monster.sql", + "0388_ambitious_micromax.sql", + "0384_funny_adam_warlock.sql", "0383_tool_review_company_expiry.sql", diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index d65d39527ac..0702f8c5336 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -45,4 +45,4 @@ export { issueRelations } from "./schema/issue_relations.js"; export { issueReferenceMentions } from "./schema/issue_reference_mentions.js"; export * from "./schema/index.js"; -export { signalDatabaseWork, subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "./work-signals.js"; +export { databaseWorkScopeKey, signalDatabaseWork, subscribeDatabaseWork, databaseWorkPending, reconcileDatabaseWork } from "./work-signals.js"; diff --git a/packages/db/src/migrations/0388_ambitious_micromax.sql b/packages/db/src/migrations/0388_ambitious_micromax.sql new file mode 100644 index 00000000000..ad828b44e2c --- /dev/null +++ b/packages/db/src/migrations/0388_ambitious_micromax.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "decisions_pending_continuation_idx" ON "decisions" USING btree ("id") WHERE "decisions"."continuation_policy" = 'wake_origin_agent' AND "decisions"."metadata" ->> 'continuationPending' = 'true' AND ("decisions"."status" IN ('expired', 'cancelled') OR ("decisions"."status" = 'decided' AND "decisions"."execution_status" IN ('succeeded', 'partial', 'failed'))); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0383_snapshot.json b/packages/db/src/migrations/meta/0388_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0383_snapshot.json rename to packages/db/src/migrations/meta/0388_snapshot.json index 3336a6bb20e..c7cd58d2a5a 100644 --- a/packages/db/src/migrations/meta/0383_snapshot.json +++ b/packages/db/src/migrations/meta/0388_snapshot.json @@ -1,6 +1,6 @@ { - "id": "410457fc-266b-4bd0-8bc0-2b14b0634d07", - "prevId": "ecec138e-5312-4ad6-b051-bd4ee9ed4e34", + "id": "4f21a636-735f-4dce-9ab8-0583f3bc7b79", + "prevId": "de27fdf8-430a-443a-9502-725f92c48892", "version": "7", "dialect": "postgresql", "tables": { @@ -22110,6 +22110,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "decision_archive_notification_outbox_delivery_idx": { + "name": "decision_archive_notification_outbox_delivery_idx", + "columns": [ + { + "expression": "last_attempt_at", + "isExpression": false, + "asc": true, + "nulls": "first" + }, + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decision_archive_notification_outbox\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -24030,6 +24058,22 @@ "method": "btree", "with": {} }, + "decisions_pending_continuation_idx": { + "name": "decisions_pending_continuation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"continuation_policy\" = 'wake_origin_agent' AND \"decisions\".\"metadata\" ->> 'continuationPending' = 'true' AND (\"decisions\".\"status\" IN ('expired', 'cancelled') OR (\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" IN ('succeeded', 'partial', 'failed')))", + "concurrently": false, + "method": "btree", + "with": {} + }, "decisions_bundle_idx": { "name": "decisions_bundle_idx", "columns": [ @@ -52856,6 +52900,44 @@ "method": "btree", "with": {} }, + "tool_action_requests_executing_deadline_idx": { + "name": "tool_action_requests_executing_deadline_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'executing'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_approved_idx": { + "name": "tool_action_requests_approved_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'approved'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -52871,6 +52953,27 @@ "method": "btree", "with": {} }, + "tool_action_requests_company_interaction_idx": { + "name": "tool_action_requests_company_interaction_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "interaction_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_issue_idx": { "name": "tool_action_requests_issue_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 4e9bf240449..68f12e89697 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2703,6 +2703,13 @@ "when": 1793136115698, "tag": "0387_boring_professor_monster", "breakpoints": true + }, + { + "idx": 388, + "version": "7", + "when": 1793137983484, + "tag": "0388_ambitious_micromax", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/decisions.ts b/packages/db/src/schema/decisions.ts index e2f668c40e6..18b18d0c704 100644 --- a/packages/db/src/schema/decisions.ts +++ b/packages/db/src/schema/decisions.ts @@ -69,6 +69,8 @@ export const decisions = pgTable( table.status, table.expiresAt, ), + pendingContinuationIdx: index("decisions_pending_continuation_idx").on(table.id) + .where(sql`${table.continuationPolicy} = 'wake_origin_agent' AND ${table.metadata} ->> 'continuationPending' = 'true' AND (${table.status} IN ('expired', 'cancelled') OR (${table.status} = 'decided' AND ${table.executionStatus} IN ('succeeded', 'partial', 'failed')))`), bundleIdx: index("decisions_bundle_idx").on(table.bundleId), originIssueIdx: index("decisions_origin_issue_idx").on(table.originIssueId), companyIdempotencyUq: uniqueIndex("decisions_company_idempotency_uq") diff --git a/packages/db/src/work-signals.test.ts b/packages/db/src/work-signals.test.ts index 13bd4b8552a..b14740bd38e 100644 --- a/packages/db/src/work-signals.test.ts +++ b/packages/db/src/work-signals.test.ts @@ -3,7 +3,7 @@ import { drizzle } from "drizzle-orm/postgres-js"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { createDb, withDedicatedDbConnection } from "./client.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./test-embedded-postgres.js"; -import { databaseWorkPending, installDatabaseWorkSignals, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js"; +import { databaseWorkScopeKey, databaseWorkPending, installDatabaseWorkSignals, reconcileDatabaseWork, signalDatabaseWork, subscribeDatabaseWork } from "./work-signals.js"; const support = await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("database work signals", () => { @@ -187,3 +187,23 @@ describe("work signal query boundaries", () => { await expect(signalDatabaseWork(connection, "bad")).rejects.toThrow("inside a transaction"); }); }); + + +describe("database work scope identity", () => { + type TestConnection = { execute: () => Promise; transaction: (callback: (tx: TestConnection) => unknown) => Promise }; + function connection(): TestConnection { + return { execute: vi.fn(async () => []), transaction: async callback => callback(connection()) }; + } + it("shares owner identity through child pools and nested transactions, but isolates independent owners", async () => { + const first = installDatabaseWorkSignals(connection()); + const second = installDatabaseWorkSignals(connection()); + const child = installDatabaseWorkSignals(connection(), first); + const key = databaseWorkScopeKey(first); + expect(databaseWorkScopeKey(second)).not.toBe(key); + expect(databaseWorkScopeKey(child)).toBe(key); + await child.transaction(async tx => { + expect(databaseWorkScopeKey(tx)).toBe(key); + await tx.transaction(async nested => { expect(databaseWorkScopeKey(nested)).toBe(key); }); + }); + }); +}); diff --git a/packages/db/src/work-signals.ts b/packages/db/src/work-signals.ts index 4abb2570a5e..cf914398e5d 100644 --- a/packages/db/src/work-signals.ts +++ b/packages/db/src/work-signals.ts @@ -5,7 +5,7 @@ import { sql } from "drizzle-orm"; // an XID, so unrelated transactions keep their existing SQL behavior. type Listener = (event: "intent" | "settled", hints?: ReadonlySet) => void; type Transaction = { topics: Set; hints: Map>; xid?: Promise; uncertain: boolean }; -type Scope = { listeners: Map>; transactions: Set; root: Connection }; +type Scope = { key: object; listeners: Map>; transactions: Set; root: Connection }; type Context = { scope: Scope; transaction?: Transaction }; type Connection = { transaction: (...args: any[]) => Promise; execute: (...args: any[]) => Promise }; const contexts = new WeakMap(); @@ -19,7 +19,7 @@ function publish(scope: Scope, topic: string, event: "intent" | "settled" = "set /** Installed once by createDb; dedicated child pools share their owner's scope. */ export function installDatabaseWorkSignals(db: T, owner?: object): T { - const scope = (owner && contexts.get(owner)?.scope) || { listeners: new Map(), transactions: new Set(), root: db }; + const scope = (owner && contexts.get(owner)?.scope) || { key: {}, listeners: new Map(), transactions: new Set(), root: db }; install(db, { scope }); return db; } @@ -112,3 +112,7 @@ export function subscribeDatabaseWork(owner: object, topic: string, listener: Li } return () => { listeners.delete(listener); }; } + +export function databaseWorkScopeKey(connection: object): object { + return contexts.get(connection)?.scope.key ?? connection; +} diff --git a/server/src/__tests__/decisions-service.test.ts b/server/src/__tests__/decisions-service.test.ts index 4b418e355f1..63719ba1ea7 100644 --- a/server/src/__tests__/decisions-service.test.ts +++ b/server/src/__tests__/decisions-service.test.ts @@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import path from "node:path"; -import { eq, inArray } from "drizzle-orm"; +import { eq, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { activityLog, @@ -11,6 +11,7 @@ import { companies, companyMemberships, createDb, + withDedicatedDbConnection, decisionEffectExecutions, decisionRetention, decisions, @@ -23,10 +24,19 @@ import { import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { attentionService } from "../services/attention.js"; import { decisionService } from "../services/decisions.js"; +import * as retentionModule from "../services/decision-retention.js"; import { hashAttentionArchiveManifest } from "../services/decision-retention.js"; import type { AttentionArchiveManifestEntry, AttentionArchiveTargetSnapshot } from "@paperclipai/shared"; -const support = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { decisionContinuationService } from "../services/decision-continuations.js"; +import { registerDecisionContinuationWork } from "../services/decision-continuation-work.js"; +import { writeDecisionChange } from "../services/decision-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_DECISIONS_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePg = support.supported ? describe : describe.skip; describePg("decisionService", () => { @@ -41,9 +51,11 @@ describePg("decisionService", () => { let decidedByUserId: string; let wakes: Array>; + let stopWorker: (() => Promise) | undefined; + beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decisions-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-decisions-"); + db = createDb(externalUrl ?? tempDb!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }, 20_000); beforeEach(async () => { @@ -69,13 +81,14 @@ describePg("decisionService", () => { }); afterEach(async () => { + await stopWorker?.(); stopWorker = undefined; delete process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE; delete process.env.PAPERCLIP_DECISIONS_RECOVERY_GRACE_MS; delete process.env.PAPERCLIP_AGENT_JWT_SECRET; await db.delete(decisionEffectExecutions); await db.delete(decisionTargetIssues); await db.delete(decisions); await db.delete(decisionRetention); await db.delete(activityLog); await db.delete(issueComments); await db.delete(issueRelations); await db.delete(heartbeatRuns); await db.delete(issues); await db.delete(agents); await db.delete(companyMemberships); await db.delete(authUsers); await db.delete(companies); }); - afterAll(async () => tempDb?.cleanup()); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); const agentActor = () => ({ type: "agent" as const, companyId, agentId, runId, source: "agent_jwt" as const, onBehalfOfUserId: originResponsibleUserId, onBehalfOfMemberships: [{ companyId, membershipRole: "member", status: "active" }] }); @@ -194,6 +207,87 @@ describePg("decisionService", () => { expect(states.find((row) => row.sourceId === unreviewedIssueId)?.archivedAt).toBeNull(); }); + it.each(["invalid_signature", "manifest_mismatch", "archive_failure"] as const)("delivers a recovered archive failure without inherited pending metadata: %s", async failure => { + const activityAt = new Date("2026-04-01T00:00:00.000Z"); + const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetIssueId, + expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }]; + const row = await service().create({ companyId, actor: agentActor(), agentId, runId, title: "Archive?", body: "Exact set", + continuationPolicy: "wake_origin_agent", options: [{ id: "archive", label: "Archive", effects: [] }], + metadata: { kind: "attention_archive_proposal", manifestHash: failure === "manifest_mismatch" ? "incorrect" : hashAttentionArchiveManifest(manifest) }, + additionalTargetSnapshots: { [`attention:review:${targetIssueId}`]: { status: "attention", assigneeAgentId: null, + assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } }); + await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "archive", + decidedByUserId, decidedAt: new Date(), updatedAt: new Date(Date.now() - 120_000), + ...(failure === "invalid_signature" ? { signedSpec: "invalid" } : {}), + metadata: { ...row.metadata, continuationPending: false } }).where(eq(decisions.id, row.id)); + const original = retentionModule.decisionRetentionService; + const factory = failure === "archive_failure" ? vi.spyOn(retentionModule, "decisionRetentionService").mockImplementation((source, options) => ({ + ...original(source, options), archiveReviewedManifest: async () => { throw new Error("archive_failure"); }, + })) : undefined; + try { + expect(await service().sweepExpired()).toMatchObject({ resumed: 1 }); + const saved = await service().get(row.id); + expect(saved).toMatchObject({ executionStatus: "failed", metadata: { archiveProposalError: failure, continuationPending: false } }); + expect(saved!.metadata.continuationDeliveredAt).toEqual(expect.any(String)); + expect(wakes).toHaveLength(1); + await service().sweepExpired(); + expect(wakes).toHaveLength(1); + } finally { factory?.mockRestore(); } + }); + + it("serializes concurrent archive completion before reporting its outcome", async () => { + await db.update(issues).set({ status: "in_review", assigneeUserId: decidedByUserId }).where(eq(issues.id, targetIssueId)); + const activityAt = new Date("2026-04-01T00:00:00.000Z"); + await db.insert(decisionRetention).values({ companyId, sourceKind: "review", sourceId: targetIssueId, sourceActivityAt: activityAt }); + const manifest: AttentionArchiveManifestEntry[] = [{ companyId, sourceKind: "review", sourceId: targetIssueId, + expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Archive" }]; + const row = await service().create({ companyId, actor: agentActor(), agentId, runId, title: "Archive?", body: "Exact set", + continuationPolicy: "wake_origin_agent", options: [{ id: "archive", label: "Archive", effects: [] }], + metadata: { kind: "attention_archive_proposal", manifestHash: hashAttentionArchiveManifest(manifest) }, + additionalTargetSnapshots: { [`attention:review:${targetIssueId}`]: { status: "attention", assigneeAgentId: null, + assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } }); + let release!: () => void; + const gate = new Promise(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)); + await withDedicatedDbConnection(db, async contender => { + let observed!: () => void; + const readStarted = new Promise(resolve => { observed = resolve; }); + let decisionReads = 0; + contender.$client.options.debug = (_id, query) => { + if (/^select/i.test(query) && query.includes('from "decisions"')) { decisionReads++; observed(); } + }; + try { + second = decisionService(contender, { wakeOriginAgent: async input => { wakes.push(input); } }) + .decide({ id: row.id, optionId: "archive", decidedByUserId, userActor: boardActor() }); + await readStarted; + await contender.execute(sql`select 1`); + await contender.execute(sql`select 1`); + expect(decisionReads).toBe(1); + release(); + const results = await Promise.all([first, second]); + expect(results.map(result => result.executionStatus)).toEqual(["succeeded", "succeeded"]); + expect(calls).toBe(1); expect(wakes).toHaveLength(1); + const [state] = await db.select().from(decisionRetention); + expect(state.archiveVersion).toBe(1); expect(state.archivedAt).not.toBeNull(); + } finally { release(); await Promise.allSettled([first, ...(second ? [second] : [])]); } + }); + } finally { release(); await Promise.allSettled([first, ...(second ? [second] : [])]); factory.mockRestore(); } + }); + it("skips strict stale targets and fails closed on intersection denial", async () => { const stale = await createCommentDecision("strict"); await db.update(issues).set({ updatedAt: new Date(Date.now() + 1_000) }).where(eq(issues.id, targetIssueId)); @@ -656,4 +750,240 @@ describePg("decisionService", () => { expect(result).toMatchObject({ status: "decided", chosenOptionId: "dismissed" }); expect(wakes).toEqual([{ companyId, agentId, issueId: originIssueId, decisionId: created.id, outcome: "decided" }]); }); + describe("continuation worker", () => { + let coordinator: ReturnType | undefined; + let draining: boolean; + const errors: unknown[] = []; + const wake = vi.fn(async (_input: Record) => {}); + const workerService = () => decisionService(db, { wakeOriginAgent: wake }); + + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + draining = false; errors.length = 0; wake.mockReset(); + stopWorker = async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists decision_ack_fail on decisions`); + await db.execute(sql`drop function if exists decision_ack_fail()`); + expect(errors).toEqual([]); + }; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + async function start(svc = workerService()) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerDecisionContinuationWork(coordinator, svc, () => !draining).ready; + } + async function quiet() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(60 * 60_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); executes.mockRestore(); + } + async function pending(id: string) { + expect((await service().get(id))?.metadata.continuationPending).toBe(true); + } + async function delivered(id: string) { + expect((await service().get(id))?.metadata.continuationPending).toBe(false); + } + const markReady = (id: string, source = db) => writeDecisionChange(source, async tx => tx.update(decisions) + .set({ status: "cancelled", metadata: { continuationPending: true } }).where(eq(decisions.id, id))); + + it("coalesces sequential tenant writes without dropping pending recovery", async () => { + const row = await createCommentDecision(); + const svc = workerService(), sweep = vi.spyOn(svc, "sweepContinuations"); + await start(svc); await quiet(); sweep.mockClear(); + await svc.cancel(row.id, { actorType: "agent", actorId: agentId }); + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(sweep).toHaveBeenCalledTimes(1); + for (let i = 0; i < 5; i++) { + const next = await createCommentDecision(); + await svc.cancel(next.id, { actorType: "agent", actorId: agentId }); + await vi.advanceTimersByTimeAsync(1000); + } + expect(sweep).toHaveBeenCalledTimes(1); + const recovery = await createCommentDecision(); + await markReady(recovery.id); + await vi.advanceTimersByTimeAsync(1000); + await pending(recovery.id); + expect(sweep).toHaveBeenCalledTimes(1); + await quiet(); await delivered(recovery.id); + expect(sweep).toHaveBeenCalledTimes(2); + expect(wake.mock.calls.filter(([input]) => input.decisionId === recovery.id)).toHaveLength(1); + }); + + it.each(["decide", "cancel", "dismiss", "expire"] as const)("reactivates from real %s and leaves delivered history quiet", async action => { + const row = await createCommentDecision(); + await start(); await quiet(); + const svc = workerService(); + if (action === "decide") await svc.decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() }); + if (action === "cancel") await svc.cancel(row.id, { actorType: "agent", actorId: agentId }); + if (action === "dismiss") await svc.dismiss(row.id, decidedByUserId, boardActor()); + if (action === "expire") { await expireDecisionNow(row.id); await svc.sweepExpired(new Date(), { includeContinuations: false }); } + await quiet(); await delivered(row.id); + expect(wake).toHaveBeenCalledTimes(1); + expect(wake.mock.calls[0][0]).toMatchObject({ decisionId: row.id, companyId, agentId, issueId: originIssueId }); + }); + + it("stays quiet while open or running, then wakes when effect completion commits", async () => { + const row = await createCommentDecision(); + await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: "yes", + decidedByUserId, metadata: { continuationPending: true } }).where(eq(decisions.id, row.id)); + await start(); await quiet(); expect(wake).not.toHaveBeenCalled(); + await workerService().decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() }); + await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); + }); + + it("coalesces simultaneous foreground replay and worker delivery without replaying effects", async () => { + const row = await createCommentDecision(); + await start(); + await Promise.all(Array.from({ length: 3 }, () => workerService().decide({ + id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor(), + }).catch(error => { if (error.message !== "decision_already_resolved") throw error; }))); + await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); + }); + + it("shares an in-flight wake with a dedicated child pool and rereads after completion", async () => { + const row = await createCommentDecision(); await markReady(row.id); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => gate); + const first = decisionContinuationService(db, wake).deliver(row.id); + const childWake = vi.fn(async () => {}); + let child: Promise; + try { + await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); + child = withDedicatedDbConnection(db, async connection => { + const attempt = decisionContinuationService(connection, childWake).deliver(row.id); + expect(attempt).toBe(first); + await attempt; + }); + await vi.advanceTimersByTimeAsync(100); + } finally { release(); } + await first; await child!; + await decisionContinuationService(db, childWake).deliver(row.id); + expect(childWake).not.toHaveBeenCalled(); await delivered(row.id); + }); + + it("reconstructs pending delivery at restart without replaying decision effects", async () => { + const row = await createCommentDecision(); + wake.mockRejectedValueOnce(new Error("wake unavailable")); + await expect(workerService().decide({ id: row.id, optionId: "yes", decidedByUserId, userActor: boardActor() })).rejects.toThrow("wake unavailable"); + await pending(row.id); await start(); await quiet(); await delivered(row.id); + expect(wake).toHaveBeenCalledTimes(2); + expect(await db.select().from(issueComments).where(eq(issueComments.issueId, targetIssueId))).toHaveLength(1); + }); + + it("retries a failed acknowledgement with the existing at-least-once wake contract", async () => { + const row = await createCommentDecision(); await start(); + await db.execute(sql`create function decision_ack_fail() returns trigger language plpgsql as $$ begin + if NEW.metadata ->> 'continuationPending' = 'false' then raise exception 'ack unavailable'; end if; + return NEW; end $$`); + await db.execute(sql`create trigger decision_ack_fail before update on decisions for each row execute function decision_ack_fail()`); + await expect(workerService().cancel(row.id, { actorType: "agent", actorId: agentId })).rejects.toMatchObject({ cause: { message: "ack unavailable" } }); + await pending(row.id); + await vi.advanceTimersByTimeAsync(60_000); + await db.execute(sql`drop trigger decision_ack_fail on decisions`); + await quiet(); await delivered(row.id); + expect(wake.mock.calls.length).toBeGreaterThan(1); + for (const error of errors) expect(error).toMatchObject({ cause: { message: "ack unavailable" } }); + errors.length = 0; + }); + + it("waits for an outer transaction, ignores rollback, and recovers the committed write", async () => { + const row = await createCommentDecision(); await start(); await quiet(); + await expect(db.transaction(async tx => { + await markReady(row.id, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(60_000); + expect(wake).not.toHaveBeenCalled(); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect((await service().get(row.id))?.status).toBe("open"); + await db.transaction(async tx => { + await markReady(row.id, tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(60_000); expect(wake).not.toHaveBeenCalled(); + }); + await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("retains a first write during drain without scanning and delivers after resume", async () => { + const row = await createCommentDecision(); await start(); await quiet(); draining = true; + await markReady(row.id); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); expect(wake).not.toHaveBeenCalled(); + draining = false; await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(1); + }); + + it("retains a failed wake through drain and resumes its retry", async () => { + const row = await createCommentDecision(); await start(); + wake.mockImplementationOnce(async () => { draining = true; throw new Error("wake unavailable"); }); + await markReady(row.id); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(30_000); expect(errors).toHaveLength(1); }); + await pending(row.id); + const reads = vi.spyOn(db, "select"); await vi.advanceTimersByTimeAsync(60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; errors.length = 0; await quiet(); await delivered(row.id); expect(wake).toHaveBeenCalledTimes(2); + }); + + it("preserves changes arriving during delivery and stops after shutdown", async () => { + const firstRow = await createCommentDecision(), secondRow = await createCommentDecision(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => { await markReady(secondRow.id); await gate; }); + await markReady(firstRow.id); + const starting = start(); + try { await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); } + finally { release(); } + await starting; await quiet(); await delivered(firstRow.id); await delivered(secondRow.id); + expect(wake).toHaveBeenCalledTimes(2); + await coordinator!.stop(); coordinator = undefined; + await vi.advanceTimersByTimeAsync(60_000); expect(wake).toHaveBeenCalledTimes(2); + }); + + it("awaits an admitted wake before completing shutdown", async () => { + const row = await createCommentDecision(), other = await createCommentDecision(); + await markReady(row.id); await markReady(other.id); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + wake.mockImplementationOnce(async () => gate); + const starting = start(); + let stopped = false; + let stopping: Promise; + try { + await vi.waitFor(() => expect(wake).toHaveBeenCalledTimes(1)); + stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(60_000); + expect(stopped).toBe(false); + } finally { release(); } + await starting; await stopping!; coordinator = undefined; + expect(wake).toHaveBeenCalledTimes(1); + const sentId = wake.mock.calls[0][0].decisionId as string; + await delivered(sentId); await pending(sentId === row.id ? other.id : row.id); expect(stopped).toBe(true); + }); + + it("pages past a failed first candidate and preserves its retry", async () => { + const seed = await createCommentDecision(); + await db.delete(decisionTargetIssues).where(eq(decisionTargetIssues.decisionId, seed.id)); + await db.delete(decisions).where(eq(decisions.id, seed.id)); + const ids = Array.from({ length: 101 }, (_, i) => `10000000-0000-4000-8000-${String(i).padStart(12, "0")}`); + await db.insert(decisions).values(ids.map(id => ({ ...seed, id, status: "cancelled", metadata: { continuationPending: true } }))); + wake.mockImplementation(async input => { if (input.decisionId === ids[0]) throw new Error("first unavailable"); }); + const recovery = workerService(); + await expect(recovery.sweepContinuations()).rejects.toThrow("first unavailable"); + expect(wake).toHaveBeenCalledTimes(100); await pending(ids.at(-1)!); + expect(await recovery.sweepContinuations()).toEqual({ retry: true }); + expect(wake).toHaveBeenCalledTimes(101); await delivered(ids.at(-1)!); await pending(ids[0]); + wake.mockImplementation(async () => {}); + await start(); await quiet(); await delivered(ids[0]); expect(wake).toHaveBeenCalledTimes(102); + }); + }); + }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 2f454b226e2..8eee51072d4 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -6,7 +6,8 @@ import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import * as idleLocalWork from "../services/idle-local-work.js"; import { idleBackupWakeMarker, markIdleBackupWakeRequired } from "../services/idle-database-backup.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; -import { decisionRetentionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; +import { decisionService, decisionRetentionService, reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; + import { companies, runDatabaseBackup, type Db } from "@paperclipai/db"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { chmodSync, existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; @@ -348,6 +349,7 @@ vi.mock("../services/index.js", () => ({ })), decisionService: vi.fn(() => ({ sweepExpired: vi.fn(async () => ({ expired: 0 })), + sweepContinuations: vi.fn(async () => ({ retry: false })), })), decisionRetentionService: vi.fn(() => ({ autoArchive: vi.fn(async () => 0), @@ -913,6 +915,10 @@ describe("startServer feedback export wiring", () => { expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.customImageExpiry)).toBe(true); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(true); + const decisions = vi.mocked(decisionService).mock.results.at(-1)!.value; + expect(decisions.sweepContinuations).toHaveBeenCalledTimes(1); + expect(decisions.sweepExpired).toHaveBeenCalledWith(expect.any(Date), { includeContinuations: false }); } finally { setIntervalSpy.mockRestore(); } @@ -949,6 +955,7 @@ describe("startServer feedback export wiring", () => { expect(routineServiceMock.tickScheduledTriggers).not.toHaveBeenCalled(); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).not.toHaveBeenCalled(); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.decisionContinuation)).toBe(false); } finally { setIntervalSpy.mockRestore(); } diff --git a/server/src/index.ts b/server/src/index.ts index 37661c37c0e..afb2b803ac9 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,5 @@ +import { registerDecisionContinuationWork } from "./services/decision-continuation-work.js"; + import { registerDecisionArchiveDeliveryWork } from "./services/decision-archive-delivery-work.js"; import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js"; @@ -1835,7 +1837,9 @@ async function startServerWithDatabaseTeardown( logger.warn({ ...toolHealthSweep }, "startup tool connection health sweep found failing connections"); } }); - await step("decision_expiry", () => decisionExecutor.sweepExpired()); + await step("decision_continuation_worker", () => registerDecisionContinuationWork(deliveryWork, decisionExecutor, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); + await step("decision_expiry", () => decisionExecutor.sweepExpired(new Date(), { includeContinuations: false })); const loginSchedule = loginCleanupSchedule(db, config.heartbeatSchedulerIntervalMs); await step("login_cleanup_workers", () => registerLoginCleanupWork(deliveryWork, [ { queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult), next: loginSchedule.adapter }, @@ -1856,7 +1860,7 @@ async function startServerWithDatabaseTeardown( trackHeartbeatSchedulerWork((async () => { if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleRoutineStatusReconcile(); - trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired().catch((err: unknown) => { + trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired(new Date(), { includeContinuations: false }).catch((err: unknown) => { logger.error({ err }, "decision expiry sweep failed"); })); trackHeartbeatSchedulerWork(runRetentionSweep().catch((err: unknown) => { diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index 044c3cbfce4..d6ff2841037 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; @@ -21,7 +21,7 @@ export function registerDeadlineWork( retryMs: task.retryMs, pausePassiveMaintenance: true, onSettled: () => { requested++; }, - run: async () => { + run: async signal => { if (!task.canRun()) { requested++; return; } if (Date.now() < earliest) { next = earliest; return; } if (Date.now() < coalesceUntil && (next === null || next > Date.now())) { @@ -30,7 +30,8 @@ export function registerDeadlineWork( } const generation = ++requested; earliest = Date.now() + task.retryMs; - await task.run(); + await task.run(signal); + if (signal.aborted) return; const deadline = await task.next(); // Back off failed or still-overdue work, not a successfully read future deadline. if (deadline === null || deadline > Date.now()) earliest = 0; diff --git a/server/src/services/decision-continuation-work.ts b/server/src/services/decision-continuation-work.ts new file mode 100644 index 00000000000..c9c5b0fbcad --- /dev/null +++ b/server/src/services/decision-continuation-work.ts @@ -0,0 +1,20 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { decisionService } from "./decisions.js"; + +export function registerDecisionContinuationWork( + coordinator: ReturnType, + service: Pick, "sweepContinuations">, + canRun: () => boolean, +) { + let next: number | null = null; + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.decisionContinuation, { + retryMs: 30_000, notificationCoalesceMs: 30_000, canRun, + run: async signal => { + const result = await service.sweepContinuations(100, signal); + next = result.retry ? Date.now() + 30_000 : null; + }, + next: async () => next, + }); +} diff --git a/server/src/services/decision-continuations.ts b/server/src/services/decision-continuations.ts new file mode 100644 index 00000000000..79c70b43969 --- /dev/null +++ b/server/src/services/decision-continuations.ts @@ -0,0 +1,66 @@ +import { and, asc, eq, gt, sql } from "drizzle-orm"; +import { databaseWorkScopeKey, decisions, type Db } from "@paperclipai/db"; + +export type DecisionContinuationWake = (input: { + companyId: string; agentId: string; issueId: string; decisionId: string; + outcome: "decided" | "expired" | "cancelled"; +}) => Promise; + +const flights = new WeakMap>>(); +const ready = sql`${decisions.continuationPolicy} = 'wake_origin_agent' + AND ${decisions.metadata} ->> 'continuationPending' = 'true' + AND (${decisions.status} IN ('expired', 'cancelled') OR + (${decisions.status} = 'decided' AND ${decisions.executionStatus} IN ('succeeded', 'partial', 'failed')))`; + +export function decisionContinuationService(db: Db, wake: DecisionContinuationWake) { + const scope = databaseWorkScopeKey(db); + let pending = flights.get(scope); + if (!pending) flights.set(scope, pending = new Map()); + const inFlight = pending; + let recoveryCursor: string | undefined; + + function deliver(id: string): Promise { + const existing = inFlight.get(id); + if (existing) return existing; + const attempt = (async () => { + const [decision] = await db.select().from(decisions).where(and(eq(decisions.id, id), ready)); + if (!decision) return; + await wake({ companyId: decision.companyId, agentId: decision.originAgentId, + issueId: decision.originIssueId, decisionId: id, + outcome: decision.status === "expired" ? "expired" : decision.status === "cancelled" ? "cancelled" : "decided" }); + await db.update(decisions).set({ + metadata: sql`coalesce(${decisions.metadata}, '{}'::jsonb) || jsonb_build_object( + 'continuationPending', false, 'continuationDeliveredAt', ${new Date().toISOString()}::text)`, + updatedAt: new Date(), + }).where(and(eq(decisions.id, id), ready)); + })().finally(() => { inFlight.delete(id); }); + inFlight.set(id, attempt); + return attempt; + } + + async function sweep(maxRows = 100, signal?: AbortSignal) { + let attempted = 0; + const errors: unknown[] = []; + while (attempted < maxRows) { + if (signal?.aborted) return { retry: true }; + const limit = Math.min(100, maxRows - attempted); + const rows = await db.select({ id: decisions.id }).from(decisions) + .where(and(ready, recoveryCursor ? gt(decisions.id, recoveryCursor) : undefined)) + .orderBy(asc(decisions.id)).limit(limit); + for (const row of rows) { + if (signal?.aborted) return { retry: true }; + try { await deliver(row.id); } catch (error) { errors.push(error); } + recoveryCursor = row.id; + } + attempted += rows.length; + if (rows.length < limit) { recoveryCursor = undefined; break; } + } + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Decision continuation delivery failed"); + if (signal?.aborted) return { retry: true }; + const remaining = await db.select({ id: decisions.id }).from(decisions).where(ready).limit(1); + return { retry: remaining.length > 0 }; + } + + return { deliver, sweep }; +} diff --git a/server/src/services/decision-work-notifications.ts b/server/src/services/decision-work-notifications.ts new file mode 100644 index 00000000000..af1ac08487f --- /dev/null +++ b/server/src/services/decision-work-notifications.ts @@ -0,0 +1,9 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export function writeDecisionChange(db: Db, write: (transaction: Db) => Promise): Promise { + return db.transaction(async transaction => { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.decisionContinuation); + return write(transaction as unknown as Db); + }); +} diff --git a/server/src/services/decisions.ts b/server/src/services/decisions.ts index c9a57063548..e37d1393b76 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, or, sql } from "drizzle-orm"; +import { and, asc, count, desc, eq, gt, gte, inArray, lte, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; -import { companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db"; +import { databaseWorkScopeKey, companyMemberships, decisionBundles, decisionEffectExecutions, decisionRetention, decisions, decisionTargetIssues, heartbeatRuns, issueRelations, issues } from "@paperclipai/db"; import { ATTENTION_SOURCE_KINDS, decisionEffectTargetIssueIds } from "@paperclipai/shared"; import type { AttentionArchiveManifestEntry, DecisionEffect, DecisionInput, DecisionOption, DecisionStatsCounts, DecisionStatsResponse } from "@paperclipai/shared"; import { conflict, forbidden, notFound, tooManyRequests, unprocessable } from "../errors.js"; @@ -14,11 +14,16 @@ import { } from "./issues.js"; import { decisionRetentionService, hashAttentionArchiveManifest } from "./decision-retention.js"; +import { decisionContinuationService, type DecisionContinuationWake } from "./decision-continuations.js"; +import { writeDecisionChange } from "./decision-work-notifications.js"; + type Snapshot = { status: string; assigneeAgentId: string | null; assigneeUserId: string | null; updatedAt: string; descendantCount?: number; descendantIds?: string[]; childCount?: number; attentionArchive?: unknown }; -type Wake = (input: { companyId: string; agentId: string; issueId: string; decisionId: string; outcome: "decided" | "expired" | "cancelled" }) => Promise; -export type DecisionServiceOptions = { wakeOriginAgent: Wake }; + +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(); @@ -118,6 +123,11 @@ function boardCanActDirectly(actor: AuthorizationActor, companyId: string) { } export function decisionService(db: Db, options: DecisionServiceOptions) { + const continuations = decisionContinuationService(db, options.wakeOriginAgent); + const scope = databaseWorkScopeKey(db); + let flights = effectFlights.get(scope); + if (!flights) effectFlights.set(scope, flights = new Map()); + const executing = flights; const authz = authorizationService(db); let targetSweepCursor: string | null = null; type CreateInput = { companyId: string; actor: AuthorizationActor; agentId: string; runId: string; bundleId?: string | null; @@ -544,7 +554,19 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { } } - async function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) { + function runEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor): Promise { + const existing = executing.get(decision.id); + if (existing) return existing; + const attempt = (async () => { + const current = await get(decision.id); + if (!current) throw notFound("Decision not found"); + return current.executionStatus === "running" ? executeDecisionEffects(current, userActor) : outcome(current.id); + })().finally(() => { executing.delete(decision.id); }); + executing.set(decision.id, attempt); + return attempt; + } + + async function executeDecisionEffects(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) { const option = decision.options.find((item) => item.id === decision.chosenOptionId); if (!option || !decision.decidedByUserId) throw unprocessable("Stored decision outcome is invalid"); const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, decision.originRunId)).then((rows) => rows[0] ?? null); @@ -555,8 +577,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { options: decision.options, targetSnapshots: decision.targetSnapshots as Record, }), decision.signedSpec)) { - await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature" } }) - .where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "invalid_signature", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } if (option.id === "archive") { @@ -566,8 +588,8 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { ); const manifestHash = manifest ? hashAttentionArchiveManifest(manifest) : null; if (!manifest || manifestHash !== metadata.manifestHash) { - await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch" } }) - .where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: "manifest_mismatch", ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } const originActor: AuthorizationActor = { @@ -588,50 +610,30 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { }); } catch (error) { const message = error instanceof Error ? error.message : "archive_proposal_failed"; - await db.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message } }) - .where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "failed", updatedAt: new Date(), metadata: { ...metadata, archiveProposalError: message, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) + .where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } } - await db.update(decisions).set({ + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: "succeeded", updatedAt: new Date(), metadata: { ...metadata, ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) }, - }).where(eq(decisions.id, decision.id)); + }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } for (let index = 0; index < option.effects.length; index += 1) await executeEffect(decision, option.effects[index]!, index, userActor, decision.decidedByUserId, run?.responsibleUserId ?? null); const rows = await db.select().from(decisionEffectExecutions).where(eq(decisionEffectExecutions.decisionId, decision.id)); const successful = rows.filter((row) => row.status === "executed").length; const status = rows.every((row) => row.status === "executed") ? "succeeded" : successful ? "partial" : "failed"; - await db.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata, - ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(eq(decisions.id, decision.id)); + await writeDecisionChange(db, async tx => tx.update(decisions).set({ executionStatus: status, updatedAt: new Date(), metadata: { ...decision.metadata, + ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.executionStatus, "running")))); return outcome(decision.id); } - async function deliverContinuation( - decision: typeof decisions.$inferSelect, - outcome: "decided" | "expired" | "cancelled", - ) { - const metadata = decision.metadata as Record; - if (decision.continuationPolicy !== "wake_origin_agent" || metadata.continuationPending !== true) return; - // Delivery is intentionally at-least-once: a crash after wakeup but before - // this acknowledgement can retry, while a crash before wakeup cannot lose - // the continuation. Heartbeat wakeups coalesce concurrent queued work. - await options.wakeOriginAgent({ companyId: decision.companyId, agentId: decision.originAgentId, - issueId: decision.originIssueId, decisionId: decision.id, outcome }); - await db.update(decisions).set({ - metadata: sql`coalesce(${decisions.metadata}, '{}'::jsonb) || jsonb_build_object( - 'continuationPending', false, - 'continuationDeliveredAt', ${new Date().toISOString()}::text - )`, - updatedAt: new Date(), - }).where(and(eq(decisions.id, decision.id), sql`${decisions.metadata} ->> 'continuationPending' = 'true'`)); - } - async function resumeDecision(decision: typeof decisions.$inferSelect, userActor: AuthorizationActor) { const result = decision.executionStatus === "running" ? await runEffects(decision, userActor) : await outcome(decision.id); - await deliverContinuation(result, "decided"); + await continuations.deliver(result.id); return outcome(decision.id); } @@ -651,25 +653,25 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { } if (current.status !== "open") throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); if (current.expiresAt <= new Date()) { - const [expired] = await db.update(decisions).set({ status: "expired", updatedAt: new Date(), metadata: { ...metadata, expiredReason: "ttl", + const [expired] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: new Date(), metadata: { ...metadata, expiredReason: "ttl", ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }) - .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"), lte(decisions.expiresAt, new Date()))).returning(); + .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"), lte(decisions.expiresAt, new Date()))).returning()); if (expired) { await logActivity(db, { companyId: expired.companyId, actorType: "system", actorId: "decision-expiry-sweeper", agentId: expired.originAgentId, runId: expired.originRunId, action: "decision.expired", entityType: "decision", entityId: expired.id, details: { expiredReason: "ttl" } }); - await deliverContinuation(expired, "expired"); + await continuations.deliver(expired.id); } throw conflict("decision_expired", { code: "decision_expired" }); } if (!current.options.some((option) => option.id === input.optionId)) throw unprocessable("Unknown optionId"); const values = input.inputValues ?? {}; for (const field of current.inputs ?? []) { const value = values[field.id] ?? ""; if (field.required && !value.trim()) throw unprocessable(`Input ${field.id} is required`); if (field.maxLength && value.length > field.maxLength) throw unprocessable(`Input ${field.id} is too long`); } - const [claimed] = await db.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values, + const [claimed] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "running", chosenOptionId: input.optionId, inputValues: values, decidedByUserId: input.decidedByUserId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...metadata, decideIdempotencyKey: input.idempotencyKey ?? null, ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}), ...(input.dismissed ? { dismissed: true, dismissReason: input.dismissReason ?? null } : {}) } }) - .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning(); + .where(and(eq(decisions.id, current.id), eq(decisions.status, "open"))).returning()); if (!claimed) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); const run = await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId }).from(heartbeatRuns).where(eq(heartbeatRuns.id, claimed.originRunId)).then((rows) => rows[0] ?? null); await logActivity(db, { companyId: claimed.companyId, actorType: "system", actorId: "decision-executor", agentId: claimed.originAgentId, runId: claimed.originRunId, @@ -683,14 +685,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { const current = await get(id); if (!current) throw notFound("Decision not found"); if (actor.actorType === "agent" && actor.actorId !== current.originAgentId) throw forbidden("Only the origin agent may cancel"); if (current.status === "cancelled") { - await deliverContinuation(current, "cancelled"); + await continuations.deliver(current.id); return (await get(id))!; } - const [updated] = await db.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata, - ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(); + const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "cancelled", updatedAt: new Date(), metadata: { ...current.metadata, + ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning()); if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); await logActivity(db, { companyId: updated.companyId, actorType: actor.actorType, actorId: actor.actorId, runId: actor.runId, action: "decision.cancelled", entityType: "decision", entityId: id }); - await deliverContinuation(updated, "cancelled"); + await continuations.deliver(updated.id); return (await get(id))!; } @@ -699,14 +701,14 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { if (!verifyDecisionSpec(spec({ id: current.id, options: current.options, targetSnapshots: current.targetSnapshots as Record }), current.signedSpec)) throw forbidden("Decision signature verification failed"); const empty = current.options.find((option) => option.effects.length === 0); if (empty) return decide({ id, optionId: empty.id, decidedByUserId: userId, userActor, dismissed: true, dismissReason: reason }); - const [updated] = await db.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId, + const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "decided", executionStatus: "succeeded", chosenOptionId: "dismissed", decidedByUserId: userId, decidedAt: new Date(), updatedAt: new Date(), metadata: { ...current.metadata, dismissed: true, dismissReason: reason ?? null, - ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning(); + ...(current.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, id), eq(decisions.status, "open"))).returning()); if (!updated) throw conflict("decision_already_resolved", { code: "decision_already_resolved" }); await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-executor", agentId: updated.originAgentId, runId: updated.originRunId, responsibleUserIdOverride: userId, action: "decision.dismissed", entityType: "decision", entityId: updated.id, details: { chosenOptionId: "dismissed", decidedByUserId: userId, dismissed: true } }); - await deliverContinuation(updated, "decided"); + await continuations.deliver(updated.id); return outcome(id); } @@ -718,7 +720,7 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { return { ...bundle, decisions: created }; }); } - async function sweepExpired(now = new Date()) { + async function sweepExpired(now = new Date(), options: { includeContinuations?: boolean } = {}) { const configuredBatchSize = Number(process.env.PAPERCLIP_DECISIONS_SWEEP_BATCH_SIZE ?? 100); const batchSize = Number.isFinite(configuredBatchSize) ? Math.max(1, Math.trunc(configuredBatchSize)) @@ -737,17 +739,9 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { const recovered = await runEffects(decision, await recoveryActor(decision.companyId, decision.decidedByUserId)); if (recovered.executionStatus === "running") continue; resumed += 1; - await deliverContinuation(recovered, "decided"); - } - const pendingContinuations = await db.select().from(decisions) - .where(and(eq(decisions.continuationPolicy, "wake_origin_agent"), - sql`${decisions.metadata} ->> 'continuationPending' = 'true'`, - or(inArray(decisions.status, ["expired", "cancelled"]), and(eq(decisions.status, "decided"), - inArray(decisions.executionStatus, ["succeeded", "partial", "failed"]))))) - .orderBy(asc(decisions.updatedAt)).limit(batchSize); - for (const decision of pendingContinuations) { - await deliverContinuation(decision, decision.status === "expired" ? "expired" : decision.status === "cancelled" ? "cancelled" : "decided"); + await continuations.deliver(recovered.id); } + if (options.includeContinuations !== false) await continuations.sweep(batchSize); const ttlRows = await db.select().from(decisions) .where(and(eq(decisions.status, "open"), lte(decisions.expiresAt, now))) .orderBy(asc(decisions.expiresAt)).limit(batchSize); @@ -774,13 +768,13 @@ export function decisionService(db: Db, options: DecisionServiceOptions) { targets.some((target) => snapshots[target.id]?.status !== "done"); if (!targetGone && !targetsCompleted && decision.expiresAt >= now) continue; const reason = targetGone ? "target_gone" : targetsCompleted ? "target_completed" : "ttl"; - const [updated] = await db.update(decisions).set({ status: "expired", updatedAt: now, metadata: { ...decision.metadata, expiredReason: reason, - ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.status, "open"))).returning(); + const [updated] = await writeDecisionChange(db, async tx => tx.update(decisions).set({ status: "expired", updatedAt: now, metadata: { ...decision.metadata, expiredReason: reason, + ...(decision.continuationPolicy === "wake_origin_agent" ? { continuationPending: true } : {}) } }).where(and(eq(decisions.id, decision.id), eq(decisions.status, "open"))).returning()); if (!updated) continue; expired += 1; await logActivity(db, { companyId: updated.companyId, actorType: "system", actorId: "decision-expiry-sweeper", agentId: updated.originAgentId, runId: updated.originRunId, action: "decision.expired", entityType: "decision", entityId: updated.id, details: { expiredReason: reason } }); - await deliverContinuation(updated, "expired"); } + await continuations.deliver(updated.id); } return { expired, resumed }; } - return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired }; + return { create, createBundle, get, list, stats, outcome, decide, cancel, dismiss, sweepExpired, sweepContinuations: continuations.sweep }; } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 230be70dcf8..31ece3af9fb 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,8 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + decisionContinuation: "decision-continuation", + decisionArchiveDelivery: "decision-archive-delivery", toolReviewProjection: "tool-review-projection",