diff --git a/doc/idle-approval-retention.md b/doc/idle-approval-retention.md new file mode 100644 index 00000000000..0675d22be09 --- /dev/null +++ b/doc/idle-approval-retention.md @@ -0,0 +1,164 @@ +# Approval retention without idle polling + +This slice removes approval retention from the recurring attention-feed sweep. +It does not migrate the other attention source kinds or establish host sleep. + +## Obligation and owner + +A pending approval in an active company is eligible for automatic archiving +90 days after its last approval activity, unless kept or already archived. +This is a presentation/attention archive: it does not approve, reject, cancel, +or delete the approval. The deadline and policy are unchanged. + +Approval rows are the source of truth, including those never viewed in the +attention UI. Scheduling only existing `decision_retention` rows would miss +unseen approvals because the old feed reader materializes those rows lazily. +`approval-retention.ts` discovers eligible approvals directly and registers +the earliest actual deadline with the existing delivery coordinator. + +The query retains the attention feed's first-linked-issue rule and consumes +`executionIssueCondition` from the issue visibility owner. Hidden, harness and +agent-conversation tasks remain excluded. A waiting external conversation with +a linked pending approval is active under that same shared visibility policy; +its chat state alone cannot hide this approval. Tests exercise the Slack case. + +Each pass visits at most 100 candidates in activity/id order. The cursor keeps +the database timestamp as text so sub-millisecond precision cannot repeat a +failed page. A failed row does +not prevent later pages from progressing. The failed durable record remains +eligible for recovery. Each candidate is re-read and locked before archiving, +so a resubmission committed while the worker waited cannot archive the new +activity using an old selection. Existing retention version, keep and archive +compare-and-set checks remain authoritative. This does not add cross-table +serializability for concurrent link/visibility/company policy changes. + +## Producer coverage + +| Supported producer | Notification boundary | +| --- | --- | +| Approval create, cancel, revision, resubmit | SQL write transaction in approval records | +| Approval resolution, including hire lifecycle | SQL write, retaining the caller's outer transaction | +| Hard budget approval and budget resolution | Existing accounting transaction | +| Formal tool approval and issue link | SQL write transaction; provider work is outside it | +| Link, unlink and bulk link | Issue approval write transaction | +| Issue visibility changes and deletion | Existing issue mutation transaction | +| Routine creation rollback cleanup | Existing transaction before cascading issue deletion | +| Company activation, pause, archive and deletion | Existing company/accounting transaction | +| Keep, manual archive, revive and accepted archive manifest | Existing retention mutation transaction | +| Retention materialization or source activity change | Signal only when inserting or changing actual state | + +Approval comments do not change approval activity under the existing policy. +Company portability imports do not write approvals; restored durable records +are reconstructed at startup. Signals belong to the application DB owner and +its dedicated child pools. Independent SQL writers or another active process +need explicit integration; this change does not claim to observe them. + +## Shared contracts and lifecycle + +The main maintenance feed excludes the approval source before collecting it. +Public attention reads keep their existing default behavior. Other attention +sources remain in the aggregate sweep until separately migrated. + +The retention service accepts the source fields it actually uses rather than +requiring fabricated full UI attention items. Its existing versioned state, +audit writes, authorization and archive-notification outbox are reused. No new +queue table is introduced. Transaction-aware notifications retain intent until +the outer transaction settles, including nested writes and rollback. + +The deadline worker uses the shared scheduler's generation and drain behavior. +Startup reconstructs pending work; idle drain retains requests without passive +SQL; resume reconciles them. Failed work retains a 30-second recovery spacing. +No eligible approval means no recurring deadline or query. Completed history, +kept/archived approvals, and inactive companies are quiet until a supported +mutation changes eligibility. Shutdown joins admitted work and stops between +candidates. Notification delivery remains owned by the separate archive worker. + +A pending-only `(updated_at, id)` index bounds deadline lookup independently of +resolved approval history. The migration uses concurrent, idempotent creation +and the existing migration runner's nontransactional execution list. + +## Verification and limits + +Real-Postgres lifecycle tests cover unseen sources, future deadlines, retained +history, keep/unkeep, archive/revive, revision/resubmission, company activity, +linked issue visibility/deletion/order, normal versus maintenance attention +reads, nested commits/rollback, startup and first-write drain, failed database +writes through drain, bounded progress past failure, shutdown, and a concurrent +resubmission. The waiting-Slack case also checks durable notification creation. + +Existing approval, attention, retention, budget, tool-gateway and startup suites +are relevant consumers. Track final results and any failures in the PR and the +[stack ledger](plans/2026-10-24-recurring-server-sleep-stack.md). + +The Cloud sleep guards are unchanged. An in-process deadline cannot wake stopped +compute, and this approval deadline has not been added to Cloud's durable wake +contract. This proves only the named worker's quiet/reactivation behavior. + + +## Initial review repair verification + +The feed and retention worker now share `approval-visibility.ts` for the first +linked issue and its execution visibility. The feed still applies actor-specific +read access before including a linked approval. Both link orders are covered: +hidden first/visible later stays excluded; visible first/hidden later stays +eligible. The worker's current-state recheck remains under the approval row lock. + +Manual keep, archive and revive materialize an unseen approval on demand inside +the existing retention write transaction. They lock the company-scoped approval +row in the same order as automatic archival, read its real activity timestamp, +and use the existing versioned `syncItems` operation. Missing or unauthorized +sources retain their existing rejection; other source kinds are unchanged. This +adds no eager startup materialization or recurring scan. + +Four real regressions failed before repair: unseen young keep/archive/revive +returned 404, and the atomic materialization test could not reach the keep write. +All pass after repair. A failing keep update rolls back the inserted retention +row and emits no audit; the next successful keep persists exactly one audit. +All 26 lifecycle cases pass, including the existing timestamp-precision failure +rotation. The complete eight-suite consumer selection passes 311 cases at +`063b61584d`: approval worker/service, attention, retention, budget, tool gateway, +startup and scoped deadlines. Server typecheck and production build pass. + +Normal parent reconciliation preserves every historical parent SQL file and all +390 parent journal entries. The approval index is regenerated through the normal +workflow as concurrent/idempotent `0392_small_frightful_four.sql`; five generated +snapshots remain (0388–0392). Schema drift, safety/numbering and actual fresh +migration apply/replay on five isolated PostgreSQL databases pass. Earlier +transaction-stub, budget-import, query-probe and CI failures remain recorded in +the stack ledger. Fresh current-head CI and both reviews are required. + + +## Wake amplification repair + +Ordinary creation, links, revision requests and resubmission signal retention +only. Tool-review queues are notified by the actual terminal transition owners: +approve/reject in approval resolution, and cancel in the approval service. These +are the same terminal transitions that notified tool reviews before this PR. +Cancellation must remain a producer because it settles parked tool actions as +unavailable. No extra wrapper or scope argument is threaded through those calls. + +The aggregate retention worker now coalesces sequential notifications for thirty +seconds after a successful reconciliation. This is a single instance-wide read +budget, stricter than granting each company an independent thirty-second budget, +and matches the former default maintenance cadence. Known business deadlines +still preempt the coalescing window. The window schedules no work on its own; +without a notification or saved deadline the worker remains quiet. Newly changed +eligibility can wait up to that window, as under the old default maintenance. + +Three new real regressions failed before repair: approve and reject cases +observed unwanted tool-review signals from ordinary mutations, and four +sequential tenant writes caused eight global reads rather than one two-read +reconciliation. Terminal approve/reject/cancel signals are now tested explicitly. +With another company's pending approval present, sequential writes cause no extra +SQL within the window and one reconciliation when the window ends. + +The first coalescing fixture incorrectly expected the pending-work hold to clear +before the requested reconciliation; it now lets the deferred wake remain owed +and measures SQL directly. The first broad repair passed 313 cases and failed +the existing parked-review cancellation case. Restoring the original cancel +producer fixed that regression without weakening its outcome assertion. All +315 cases in eight full suites now pass at `dd844f4e8c`, including 30 approval +worker cases; server types/build pass. The earlier security check at `5991f257bb` +was action-required and Greptile reported service error reference +`8fb1835c-e564-476b-b802-d5be164ea915`. Neither is claimed as passing; the repaired +head requires fresh security, CI and both bot reviews. 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 8d5a064676e..c2f604bfcc6 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -15,7 +15,7 @@ Do not mark an item done from a timer removal or a green check alone. | 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | [#17443](https://github.com/paperclipai/paperclip/pull/17443), 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; archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged after exact-head 5/5 reviews and passing current CI | +| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Approval retention implemented in slice 14; final review/CI pending; other source 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) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) merged | @@ -32,7 +32,8 @@ Do not mark an item done from a timer removal or a green check alone. | 14 | Skill-version retention and native-harness backup expiration | Separate retention owners and existing retention guarantees | Pending | | 15 | Automatic database backups | Shared scheduling, actual backup deadlines, existing checkpoint contract | Pending | | 16 | Aggregate maintenance tick | Retire enabled and disabled branches after all constituent owners migrate | Pending | -| 17 | Grok routine dispatch | Committed reservations and feature changes; prepared-only recovery | Slice 13 implemented, local verification in progress | +| 17 | Grok routine dispatch | Committed reservations and feature changes; prepared-only recovery | [#17490](https://github.com/paperclipai/paperclip/pull/17490) merged after exact-head 5/5 reviews and passing CI | +| 18 | Abandoned import transfer spools | Creation/completion events, existing retention deadline, startup recovery | Pending; explicitly retained from the remaining-blocker checklist | ## Stack and verification contract @@ -770,6 +771,7 @@ files. All runtime and acceptance test sources are byte-identical to its tested `85ce6545bf` tree (248 cases, server types/build). Fresh final-head CI and reviews remain required after retargeting to master. + ## Slice 13: Grok routine dispatch The one-second idle scan is replaced with a shared-coordinator worker. The existing @@ -806,3 +808,221 @@ independent Runner E2E accounting files. All runtime and acceptance sources matc its tested/reviewed `679f523d78` tree; the 273-case, type/build and migration receipts retain their original revisions. Migration 0391 and all parent journal entries are unchanged. Fresh final-head CI and reviews required against master. + +## Next owner audit: decision retention + +Slice 13 is [#17490](https://github.com/paperclipai/paperclip/pull/17490), head +`706beb01f2`, based on the repaired target PR #17486 (`1bbfc69590`). It is +attached and review was requested. The target PR now has fresh Greptile 5/5 +with its mixed-hint finding resolved; its fresh CI is still running. Grok's CI +and review are running. No new readiness or merge claim is made. + +Branch `codex/sleep-14-decision-retention` starts from #17490. This section is +an audit, not an implemented retention migration. The original full inventory +remains the completion scope. + +The existing startup/tick retention function selects every active company, +builds `attentionService.list({includeDismissed:true, all:true, +allowUnscopedAll:true})`, then calls `autoArchive`. Notification delivery was +already separated in slice 08. Retention itself is not just an expiry scan of +`decision_retention`: `attentionService` materializes those records via +`syncItems` while building the feed. An item never viewed in the UI still needs +retention coverage. Removing the periodic feed build and looking only at existing +retention rows would silently miss such sources. + +The auto-archive policy uses 90 days since source activity, `keep = false`, and +no existing archive. The 30-day shelf default and per-queue shelf overrides are +UI classification, not the automatic archive deadline. Keep these concepts +separate. Preserve version CAS, actor columns, archiveVersion, notification +outbox creation, and audit in the owning transaction. Manual archive/revive, +keep/unkeep, and archive-proposal manifests are also producers of schedule +changes; they must not require another unrelated source write to take effect. + +Current feed sources and eligibility dependencies observed in attention.ts: + +| Source | Activity timestamp / eligibility dependencies to audit | +| --- | --- | +| approval | Approval updatedAt; pending status; issue approval links | +| issue_thread_interaction | Interaction updatedAt; pending status; issue and audience policy | +| decision | Decision updatedAt; open status; current feed selection has a default 500-row cap | +| join_request | Join updatedAt; pending_approval; matching company invite | +| recovery_action | Recovery updatedAt; active/escalated; human owner and visible source issue | +| blocker_attention | Prospective blockedTransitionAt or terminal blocker's updatedAt; dependency graph, liveness, monitor and unblock-owner state | +| review | Issue updatedAt; in_review; human participant/assignee, pending approval, or stalled review liveness | +| failed_run | finishedAt/updatedAt/createdAt; bounded-retry-exhaustion event; failed/timed_out; nonterminated agent; no newer run for that agent/task | +| budget_alert | Incident updatedAt; current budget overview/window; hard threshold or observed >=85% | +| agent_error_alert | Agent updatedAt; error status | + +`productivity_review` remains in the source-kind contract and retention origin +resolver, but this inspected feed does not directly emit that kind. Do not invent +a new producer or assume all enum values imply live rows. Failed-run attention +has no simple age cutoff in `listAttentionExhaustedRuns`; old rows cannot be +excluded merely because they are old. + +Review/blocker attention depends on more than issue status. The review liveness +input includes agents, queued/running runs, wake requests, pending interactions, +approvals, external-response waits, recovery paths, monitors, and `now`. Identify +any clock-driven eligibility transitions as well as source mutations before +removing the backstop. Company activation must discover old eligible sources, +while company deactivation must stop its passive work. + +A company-scoped dirty signal can coalesce a source change into a finite feed +reconciliation, but it is only valid after every supported source writer and +transaction owner is accounted for. A live-event callback or activity-log hook +alone is not a proven durable producer contract. Do not hide provider/domain +policy in the shared scheduler or add blanket database-write interception. + +Also avoid a notification feedback loop: syncItems currently attempts an +on-conflict insert for all items on every feed read, then updates changed activity +versions. Unconditionally signaling every such attempted insert would cause the +worker's own read/materialization to continually wake itself. Any notification +at this boundary must distinguish actual new/changed retention state and converge. + +Required evidence must include an unseen source aging into retention, all audited +source classes and activation paths, exact archive deadline, activity moving a +deadline later, keep/unkeep and manual archive/revive, no repeated feed rebuild +with unchanged pending or completed history, failure/restart/drain, and a source +change during archive selection. The current version CAS protects retention-row +changes; it does not by itself prove that an underlying source did not change +between the feed read and archive. Preserve or strengthen that boundary explicitly +without silently archiving freshly active work. No retention code or PR is yet +claimed for this branch. + + +Latest remote verification: #17486 has fresh Greptile 5/5 at `1bbfc69590`, +all review findings resolved, and one live build job still pending in run +40132295156. #17490 has fresh Greptile 5/5 at `706beb01f2`, no review threads, +and active CI run 40132581543 (24 completed jobs, 22 still pending, no failures +at this snapshot). These are verified live waits, not completed checks. Do not +rerun them merely because a later observation times out. Continue inspecting +these runs while implementing retention, and propagate any required parent +repair into the later branch. No stack PR was merged. + + +## Approval retention implementation in progress + +Remote verification on October 24: #17486 at `1bbfc69590` and #17490 at +`706beb01f2` each have 56 completed checks, all successful or accepted +neutral/skipped outcomes. Their PR bodies now record completion. Neither was +merged; this remains a stack. + +The retention audit is now split by source owner. Slice 14 migrates **approval +retention only**. Decisions, interactions, join requests, recovery actions, +blocker attention, review, failed runs, budget alerts and agent errors remain +explicitly pending. The original full recurring-work inventory remains scope. + +The implementation discovers pending approvals directly, including approvals +never materialized by an attention-feed read. It uses active-company status, +the feed's first linked issue, shared execution visibility, keep/archive state, +and the approval's updated timestamp. The next obligation is updated timestamp +plus 90 days. Each bounded pass rechecks and locks the approval before using +existing versioned retention, audit and notification-outbox writes. Other +retention sources still use the aggregate sweep; approval collection is excluded +from that maintenance feed before its query, while normal UI reads are unchanged. + +Signals cover approval create/cancel/revision/resubmit/resolve, hard budget +approvals, formal tool approvals, link/unlink/bulk linking, issue visibility and +cascading deletion, company activation/pause/archive/deletion, and keep/archive/ +revive/accepted archive manifests. Source comments do not alter approval activity +under the existing policy. Company portability imports no approval records; +offline restore is reconstructed at startup. Independent SQL writers still +require a wake integration. The Cloud sleep proof is unchanged. + +Initial real-Postgres proof: 14 lifecycle cases passed, then 19 including actual +archive-write faults, page progress past a failed row, startup drain, first-link +ordering and shutdown. The first consumer run passed retention and startup but +failed five approval unit cases because its transaction stub omitted nested +transactions; the stub was repaired without weakening production behavior. +The budget suite initially skipped due to embedded-Postgres availability. Budget +and attention suites now accept validated disposable external test databases. +Expanded consumer verification, final index/migration proof, build and PR review +remain required. No slice-14 PR has been published yet. + + +Slice 14 expanded proof: all 20 worker lifecycle cases pass. The shared execution +visibility predicate is used directly, and a waiting Slack conversation with a +pending approval remains eligible with its archive notification persisted. The +repaired approval stub and attention/retention/startup suites pass. Of 96 consumer +cases, one budget case initially timed out during lazy lifecycle module loading +and raced teardown; moving the import into setup preserves the original timeout +and all five budget cases pass. No assertion or production behavior was weakened. +The initial index diagnostic failed to construct a Drizzle client around the +scoped raw driver; supplying its existing driver options fixed the diagnostic. +The actual deadline query with 40,000 resolved approvals uses the pending index, +four shared buffer hits and 0.314 ms execution; test data rolled back. Migration +replay and checks pass. Workspace types, build and tool-gateway tests remain live +in the current verification run. This is not final acceptance. + + +Slice 14 final local verification passes 317 selected tests: 20 new lifecycle, +96 approval/attention/retention/startup/budget consumers (including the repaired +budget case), and 201 tool-gateway cases. Workspace typecheck, production build, +migration numbering/safety, fresh replay and idempotent index replay pass. The +implementation is committed as `ea3dd40cdd`; remote publication, CI and review +remain required. The earlier failures remain recorded above. + + +## Slice 14 publication and precision review repair + +Approval retention is [#17497](https://github.com/paperclipai/paperclip/pull/17497), +initial head `2f334c678b`, based on #17490. It is attached. Initial CI run +40135489716 completed without a failed CI job. Greptile gave 4/5 with one P2: +JavaScript Date truncation can repeat a full failed page whose source timestamp +contains microseconds (comment 4368633789, thread PRRT_kwDORclPyM6v79u7). + +A database-generated timestamp regression failed before repair: all 100 failed +rows repeated and blocked row 101. The cursor now preserves the database text +timestamp. All 21 lifecycle cases pass afterward, and server typecheck passes. +During the full regression run, the earlier quiet-window test crossed its real +deadline after settlement. It now measures an explicit window before that +deadline; it still asserts no SQL and eventual archival. Total selected coverage +is now 318 unique tests. Fresh remote checks/review are required after this fix. + +Related precision audit is required before final stack acceptance: the earlier +TTL worker (#17479) and effect-recovery worker (#17482) retain Date cursors too. +Add database-precision regressions and repair those owning PRs, then propagate +parent updates through the stack. The already-merged terminal-workspace sweep +also uses a Date keyset cursor; verify and address it as a focused follow-up if +confirmed. These are concrete remaining verification duties, not waived by the +earlier green checks. + +Branch `codex/sleep-15-join-retention` exists at the initial slice-14 head, with +no implementation yet. Its producer audit found request writes in access routes +(accept, replay, auto-approve, approve, reject, key claim), Slack identity access +creation, and company deletion. Eligibility is pending_approval plus matching +invite/company and active company. Invite expiry/revocation do not remove the +existing feed item. Preserve that policy. Retention controls and company status +changes must notify this owner. Decision retention remains separately pending; +its visibility dependencies differ, and normal decisions already have a +maximum 30-day TTL before the 90-day archive threshold. + + +Slice 14 current review repairs: share first-linked-issue/execution visibility in +one owner used by the feed and worker; keep actor read filtering. Materialize +unseen approvals inside existing manual retention transactions, locking the +source before state mutation. Four real failures (three 404s and atomic keep +rollback) are repaired; the reverse visibility-order regression also passes. +All 26 lifecycle cases and all 311 cases across eight complete consumer suites, +server types/build, schema drift, migration safety and fresh apply/replay pass. +Index 0392 is generated after all 390 unchanged parent journal entries; five +snapshots remain. Current parent is `679f523d78`; fresh head gates remain required. + + +Slice 14 wake-amplification follow-up: ordinary approval/link/revision/resubmit +writes no longer notify tool reviews. Approve/reject/cancel retain their existing +terminal wake contract. Retention uses a thirty-second aggregate notification +budget, with known deadlines preempting and no idle recurring scan. Three real +regressions failed before repair. A broad run also caught a removed cancellation +producer (313 pass/1 fail); it is restored. All 315 cases in eight complete suites, +including 30 worker cases, and server types/build pass at `dd844f4e8c`. Prior +coalescing-fixture, security and Greptile service failures remain in the owning +audit. Fresh exact-head checks and reviews are required on publication. + + +Slice 13 merged `15cddfcb0c` → `2adcde162e` after exact-head Greptile/Ponytail +5/5 and all 58 check contexts passed or were neutral/skipped. Slice 14 normally +integrates that landed dependency, preserving four independent Runner E2E +accounting files. Its entire runtime and acceptance sources match the tested +and reviewed `3efda33812` tree (315 cases, server types/build). Migration 0392 +and the parent journal prefix are unchanged. Fresh final-head CI/reviews are +required against master. The abandoned-import-spool inventory duty remains. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 0b743db6fce..2b987eb9720 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -510,6 +510,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0389_brave_hulk.sql", "0390_salty_killraven.sql", "0391_third_roulette.sql", + "0392_small_frightful_four.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0392_small_frightful_four.sql b/packages/db/src/migrations/0392_small_frightful_four.sql new file mode 100644 index 00000000000..b8933f13c11 --- /dev/null +++ b/packages/db/src/migrations/0392_small_frightful_four.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "approvals_pending_retention_idx" ON "approvals" USING btree ("updated_at","id") WHERE "approvals"."status" = 'pending'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0387_snapshot.json b/packages/db/src/migrations/meta/0392_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0387_snapshot.json rename to packages/db/src/migrations/meta/0392_snapshot.json index 7fd3f6cbe93..0f1a1921bbb 100644 --- a/packages/db/src/migrations/meta/0387_snapshot.json +++ b/packages/db/src/migrations/meta/0392_snapshot.json @@ -1,6 +1,6 @@ { - "id": "de27fdf8-430a-443a-9502-725f92c48892", - "prevId": "da3229b5-4eab-4d8b-ba6b-d756e6c42430", + "id": "6974dd25-b8e9-4143-b4b7-ee5a5c3ff54a", + "prevId": "e080fc1b-af8e-4012-8ce5-3b5065b65143", "version": "7", "dialect": "postgresql", "tables": { @@ -5721,6 +5721,28 @@ } }, "indexes": { + "approvals_pending_retention_idx": { + "name": "approvals_pending_retention_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"approvals\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "approvals_company_status_type_idx": { "name": "approvals_company_status_type_idx", "columns": [ @@ -24058,6 +24080,66 @@ "method": "btree", "with": {} }, + "decisions_running_recovery_idx": { + "name": "decisions_running_recovery_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" = 'running'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "decisions_open_expiry_idx": { + "name": "decisions_open_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"status\" = 'open'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "decisions_pending_continuation_idx": { + "name": "decisions_pending_continuation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"decisions\".\"continuation_policy\" = 'wake_origin_agent' AND \"decisions\".\"metadata\" ->> 'continuationPending' = 'true' AND (\"decisions\".\"status\" IN ('expired', 'cancelled') OR (\"decisions\".\"status\" = 'decided' AND \"decisions\".\"execution_status\" IN ('succeeded', 'partial', 'failed')))", + "concurrently": false, + "method": "btree", + "with": {} + }, "decisions_bundle_idx": { "name": "decisions_bundle_idx", "columns": [ @@ -32047,6 +32129,28 @@ "method": "btree", "with": {} }, + "grok_dispatch_prepared_idx": { + "name": "grok_dispatch_prepared_idx", + "columns": [ + { + "expression": "created_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"grok_bot_dispatches\".\"status\" = 'prepared'", + "concurrently": false, + "method": "btree", + "with": {} + }, "grok_dispatch_company_status_idx": { "name": "grok_dispatch_company_status_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index c75fc030ee5..8536f384cb0 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2731,6 +2731,13 @@ "when": 1793142971791, "tag": "0391_third_roulette", "breakpoints": true + }, + { + "idx": 392, + "version": "7", + "when": 1793143372644, + "tag": "0392_small_frightful_four", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/approvals.ts b/packages/db/src/schema/approvals.ts index f4cd512c1fa..052e37b03a4 100644 --- a/packages/db/src/schema/approvals.ts +++ b/packages/db/src/schema/approvals.ts @@ -1,3 +1,4 @@ +import { sql } from "drizzle-orm"; import { pgTable, uuid, text, timestamp, jsonb, index } from "drizzle-orm/pg-core"; import { companies } from "./companies.js"; import { agents } from "./agents.js"; @@ -19,6 +20,7 @@ export const approvals = pgTable( updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(), }, (table) => ({ + pendingRetentionIdx: index("approvals_pending_retention_idx").on(table.updatedAt, table.id).where(sql`${table.status} = 'pending'`), companyStatusTypeIdx: index("approvals_company_status_type_idx").on( table.companyId, table.status, diff --git a/server/src/__tests__/approval-retention-work.test.ts b/server/src/__tests__/approval-retention-work.test.ts new file mode 100644 index 00000000000..9315e401089 --- /dev/null +++ b/server/src/__tests__/approval-retention-work.test.ts @@ -0,0 +1,423 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, approvals, chatConversations, chatEndpoints, companies, createDb, subscribeDatabaseWork, decisionArchiveNotificationOutbox, decisionRetention, issueApprovals, issues, toolApplications, toolConnections, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { approvalRecords } from "../services/approvals.js"; +import { approvalRetentionService, registerApprovalRetentionWork } from "../services/approval-retention.js"; +import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "../services/decision-retention.js"; +import { attentionService } from "../services/attention.js"; +import { issueApprovalService } from "../services/issue-approvals.js"; +import { companyService } from "../services/companies.js"; +import { issueService } from "../services/issues.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_APPROVAL_RETENTION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; +(support.supported ? describe : describe.skip)("approval retention worker", () => { + let temporary: Awaited> | undefined; + let db: ReturnType, companyId: string; + let coordinator: ReturnType | undefined; + let draining: boolean; + const errors: unknown[] = []; + beforeAll(async () => { + if (!externalUrl) temporary = await startEmbeddedPostgresTestDatabase("approval-retention-"); + db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + companyId = randomUUID(); draining = false; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Approval retention", issuePrefix: `A${companyId.slice(0, 6)}` }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists approval_retention_fail on decision_retention`); + await db.execute(sql`drop function if exists approval_retention_fail()`); + await db.delete(decisionArchiveNotificationOutbox); await db.delete(decisionRetention); + await db.delete(activityLog); await db.delete(issueApprovals); await db.delete(approvals); + await db.delete(chatConversations); await db.delete(chatEndpoints); + await db.delete(toolConnections); await db.delete(toolApplications); + await db.delete(issues); await db.delete(agents); await db.delete(companies); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + const create = (at = new Date(), database: Db = db) => approvalRecords(database).create(companyId, { + type: "request_board_approval", payload: {}, createdAt: at, updatedAt: at, + }); + const old = () => new Date(Date.now() - delay - 60_000); + const state = (id: string) => decisionRetentionService(db).getState(companyId, "approval", id); + const authActor = () => ({ type: "board" as const, source: "local_implicit" as const, userId: "board-user", companyIds: [companyId], isInstanceAdmin: true }); + const actor = { actorType: "user" as const, actorId: "board-user", agentId: null, userId: "board-user", runId: null, agentApiKeyId: null, responsibleUserId: "board-user" }; + const mutation = (id: string) => ({ companyId, sourceKind: "approval" as const, sourceId: id, authActor: authActor(), actor }); + async function start(service = approvalRetentionService(db)) { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, onError: error => errors.push(error) }); + await registerApprovalRetentionWork(coordinator, service, () => !draining).ready; + return service; + } + async function settle() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + } + async function quiet() { + await settle(); + expect(coordinator!.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(300_000); + expect(reads).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); executes.mockRestore(); writes.mockRestore(); + } + async function archived(id: string) { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect((await state(id))?.archivedAt).toBeInstanceOf(Date); + }, { timeout: 5000 }); + } + async function issue(hiddenAt: Date | null = null) { + const [row] = await db.insert(issues).values({ companyId, title: "Approval target", status: "todo", hiddenAt }).returning(); + return row; + } + + it("is quiet with no approvals and archives unseen durable approvals at their actual deadline", async () => { + await start(); await quiet(); + const deadline = Date.now() + 600_000; + const row = await create(new Date(deadline - delay)); + await settle(); + expect(await state(row.id)).toBeNull(); + expect(coordinator!.nextWakeAt()).toBe(deadline); + const quietWindow = deadline - Date.now() - 30_000; + expect(quietWindow).toBeGreaterThan(0); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(quietWindow); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await vi.advanceTimersByTimeAsync(60_000); + await archived(row.id); await quiet(); + expect((await db.select().from(approvals).where(eq(approvals.id, row.id)))[0].status).toBe("pending"); + }); + + it.each(["approved", "rejected", "cancelled"] as const)("wakes parked tool reviews only on terminal %s transition", async outcome => { + const queues = [DELIVERY_QUEUES.toolReviewExpiry, DELIVERY_QUEUES.toolReviewRecovery, + DELIVERY_QUEUES.toolReviewInterruption, DELIVERY_QUEUES.toolReviewProjection]; + const events = queues.map(() => [] as string[]); + const subscriptions = queues.map((queue, index) => subscribeDatabaseWork(db, queue, event => { events[index].push(event); })); + try { + const row = await create(), target = await issue(); + for (let pass = 0; pass < 3; pass++) { + await issueApprovalService(db).link(target.id, row.id); + await issueApprovalService(db).unlink(target.id, row.id); + } + await approvalRecords(db).requestRevision(row.id, "board-user"); + await approvalRecords(db).resubmit(row.id); + expect(events).toEqual(queues.map(() => [])); + if (outcome === "cancelled") await approvalRecords(db).cancel(row.id, "cancelled by user"); + else await approvalRecords(db).resolveApproval(row.id, outcome, "board-user", null); + expect(events).toEqual(queues.map(() => ["intent", "settled"])); + } finally { for (const unsubscribe of subscriptions) unsubscribe(); } + }); + + it("bounds aggregate retention reads to one reconciliation per thirty seconds of sequential tenant writes", async () => { + const [otherCompany] = await db.insert(companies).values({ name: "Other tenant", issuePrefix: `B${companyId.slice(0, 6)}` }).returning(); + const other = await approvalRecords(db).create(otherCompany.id, { type: "request_board_approval", payload: {} }); + const service = await start(); await settle(); + const sweep = vi.spyOn(service, "sweep"), reads = vi.spyOn(db, "select"); + await create(); + await vi.waitFor(() => expect(sweep).toHaveBeenCalledTimes(1)); + await sweep.mock.results[0].value; + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const afterFirst = reads.mock.calls.length; + expect(afterFirst).toBe(2); + for (let pass = 0; pass < 3; pass++) { + await create(); + await vi.advanceTimersByTimeAsync(100); + } + await vi.advanceTimersByTimeAsync(29_000); + expect(reads).toHaveBeenCalledTimes(afterFirst); + await vi.advanceTimersByTimeAsync(2000); + await vi.waitFor(() => expect(reads).toHaveBeenCalledTimes(afterFirst + 2)); + reads.mockRestore(); + expect((await db.select().from(approvals).where(eq(approvals.id, other.id)))[0].status).toBe("pending"); + }); + + it("recovers old unseen approvals on startup and leaves completed history quiet", async () => { + const due = await create(old()), completed = await create(old()); + await approvalRecords(db).resolveApproval(completed.id, "approved", "board-user", null); + await start(); await archived(due.id); await quiet(); + expect(await state(completed.id)).toBeNull(); + }); + + it("disarms on revision and moves the deadline when resubmitted", async () => { + await start(); const row = await create(); await settle(); + await approvalRecords(db).requestRevision(row.id, "board-user"); await quiet(); + const resubmitted = await approvalRecords(db).resubmit(row.id); await settle(); + expect(coordinator!.nextWakeAt()).toBe(resubmitted.updatedAt.getTime() + delay); + await approvalRecords(db).cancel(row.id); await quiet(); + }); + + it.each(["keep", "archive", "revive"] as const)("applies %s to an unseen young approval without a feed read", async action => { + const row = await create(); + expect(await state(row.id)).toBeNull(); + const retention = decisionRetentionService(db); + const result = action === "keep" ? await retention.setKeep({ ...mutation(row.id), keep: true }) + : action === "archive" ? await retention.archive(mutation(row.id)) : await retention.revive(mutation(row.id)); + expect(result).toMatchObject({ sourceKind: "approval", sourceId: row.id, sourceActivityAt: row.updatedAt, keep: action === "keep" }); + expect(result.archivedAt).toEqual(action === "archive" ? expect.any(Date) : null); + expect(await state(row.id)).toMatchObject({ id: result.id, keep: action === "keep" }); + }); + + it("rolls back on-demand materialization when the manual keep write fails", async () => { + const row = await create(); + await db.execute(sql`create function approval_retention_fail() returns trigger language plpgsql as $$ begin + if NEW.keep then raise exception 'manual keep unavailable'; end if; return NEW; end $$`); + await db.execute(sql`create trigger approval_retention_fail before update on decision_retention for each row execute function approval_retention_fail()`); + await expect(decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true })) + .rejects.toMatchObject({ cause: { message: "manual keep unavailable" } }); + expect(await state(row.id)).toBeNull(); + expect(await db.select().from(activityLog).where(eq(activityLog.action, "decision_retention.keep_updated"))).toHaveLength(0); + await db.execute(sql`drop trigger approval_retention_fail on decision_retention`); + expect(await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true })).toMatchObject({ keep: true }); + expect(await db.select().from(activityLog).where(eq(activityLog.action, "decision_retention.keep_updated"))).toHaveLength(1); + }); + + it("honors keep and wakes when an old approval is unkept", async () => { + const row = await create(old()); + await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: true }); + await start(); await quiet(); expect((await state(row.id))?.archivedAt).toBeNull(); + await decisionRetentionService(db).setKeep({ ...mutation(row.id), keep: false }); + await archived(row.id); await quiet(); + }); + + it("disarms on manual archive and reactivates on revive", async () => { + const row = await create(old()); + await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + await decisionRetentionService(db).archive(mutation(row.id)); + await start(); await quiet(); + await decisionRetentionService(db).revive(mutation(row.id)); + await archived(row.id); await quiet(); + expect((await state(row.id))?.archivedReason).toBe("idle_ttl"); + }); + + it("waits for inactive companies and wakes on company activation", async () => { + const row = await create(old()); + await companyService(db).update(companyId, { status: "paused" }); + await start(); await quiet(); expect(await state(row.id)).toBeNull(); + await companyService(db).update(companyId, { status: "active" }); + await archived(row.id); await quiet(); + }); + + it("excludes hidden linked targets and wakes after unlink", async () => { + const row = await create(old()), target = await issue(new Date()); + await issueApprovalService(db).link(target.id, row.id); + await start(); await quiet(); expect(await state(row.id)).toBeNull(); + await issueApprovalService(db).unlink(target.id, row.id); + await archived(row.id); await quiet(); + }); + + it("wakes when the linked target becomes visible", async () => { + const row = await create(old()), target = await issue(new Date()); + await issueApprovalService(db).link(target.id, row.id); + await start(); await quiet(); + await issueService(db).update(target.id, { hiddenAt: null }); + await archived(row.id); await quiet(); + }); + + it("wakes after deletion cascades the excluded target link", async () => { + const row = await create(old()), target = await issue(new Date()); + await issueApprovalService(db).link(target.id, row.id); + await start(); await quiet(); + await issueService(db).remove(target.id); + await archived(row.id); await quiet(); + }); + + it("preserves normal attention reads while excluding approval retention maintenance", async () => { + const row = await create(old()); + const excluded = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, excludeSources: ["approval"] }); + expect(excluded.items.some(item => item.sourceKind === "approval")).toBe(false); + expect(await state(row.id)).toBeNull(); + const visible = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + expect(visible.items.some(item => item.subject.id === row.id)).toBe(true); + expect(await state(row.id)).not.toBeNull(); + }); + + it("does not publish work before an outer transaction commits", async () => { + await start(); await quiet(); + let id = ""; + await db.transaction(async tx => { + id = (await create(old(), tx as unknown as Db)).id; + await vi.advanceTimersByTimeAsync(60_000); + expect(await state(id)).toBeNull(); + }); + await archived(id); await quiet(); + }); + + it("discharges rolled back nested producers without archiving", async () => { + await start(); await quiet(); + await expect(db.transaction(async tx => { + await create(old(), tx as unknown as Db); throw new Error("rollback"); + })).rejects.toThrow("rollback"); + await quiet(); expect(await db.select().from(approvals)).toEqual([]); + }); + + it("retains the first producer during drain and processes it after resume", async () => { + await start(); await quiet(); draining = true; + const row = await create(old()); + await vi.advanceTimersByTimeAsync(90_000); expect(await state(row.id)).toBeNull(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; await archived(row.id); await quiet(); + }); + + it("rechecks current source activity after waiting on an approval writer", async () => { + const row = await create(old()); + let unlock!: () => void, locked!: () => void; + const barrier = new Promise(resolve => { unlock = resolve; }); + const admitted = new Promise(resolve => { locked = resolve; }); + const write = db.transaction(async tx => { + await tx.execute(sql`select id from approvals where id = ${row.id} for update`); + locked(); await barrier; + await approvalRecords(tx as unknown as Db).requestRevision(row.id, "board-user"); + await approvalRecords(tx as unknown as Db).resubmit(row.id); + }); + await admitted; + const sweep = approvalRetentionService(db).sweep(); + await vi.advanceTimersByTimeAsync(100); + unlock(); await write; await sweep; + expect(await state(row.id)).toBeNull(); + }); + + async function failArchives() { + await db.execute(sql`create function approval_retention_fail() returns trigger language plpgsql as $$ + begin + if new.archived_at is not null and exists ( + select 1 from approvals where id::text = new.source_id and payload->>'retentionTestFailure' = 'true' + ) then raise exception 'injected approval archive failure'; end if; + return new; + end $$`); + await db.execute(sql`create trigger approval_retention_fail before update on decision_retention + for each row execute function approval_retention_fail()`); + } + + it("retains failed durable work across drain and resumes recovery", async () => { + const row = await approvalRecords(db).create(companyId, { type: "request_board_approval", + payload: { retentionTestFailure: true }, updatedAt: old() }); + await failArchives(); + await start(); + expect(errors.length).toBeGreaterThan(0); errors.length = 0; + draining = true; + await vi.advanceTimersByTimeAsync(60_000); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await db.execute(sql`drop trigger approval_retention_fail on decision_retention`); + draining = false; await archived(row.id); await quiet(); + }); + + it("advances bounded pages past a failed row without losing its retry", async () => { + const at = old(); + await db.insert(approvals).values(Array.from({ length: 101 }, (_, i) => ({ companyId, + type: "request_board_approval", payload: { retentionTestFailure: i === 0 }, + createdAt: at, updatedAt: new Date(at.getTime() + i), + }))); + await failArchives(); + const service = approvalRetentionService(db); + await expect(service.sweep()).rejects.toThrow("Failed query"); + expect(await service.sweep()).toEqual({ archived: 1 }); + expect((await db.select().from(decisionRetention)).filter(row => row.archivedAt)).toHaveLength(100); + expect(await service.next()).not.toBeNull(); + await db.execute(sql`drop trigger approval_retention_fail on decision_retention`); + expect(await service.sweep()).toEqual({ archived: 1 }); expect(await service.next()).toBeNull(); + }); + + it("reconstructs startup obligations while paused and performs no passive SQL", async () => { + const row = await create(old()); draining = true; + await start(); + const reads = vi.spyOn(db, "select"); + await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(await state(row.id)).toBeNull(); + draining = false; await archived(row.id); await quiet(); + }); + + it("matches the feed's first linked issue rather than any visible link", async () => { + const row = await create(old()), one = await issue(), two = await issue(); + const [first, second] = [one, two].sort((a, b) => a.id.localeCompare(b.id)); + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, first.id)); + await issueApprovalService(db).linkManyForApproval(row.id, [second.id, first.id]); + expect((await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true })).items + .some(item => item.subject.id === row.id)).toBe(false); + await start(); await quiet(); + await issueApprovalService(db).unlink(first.id, row.id); + await archived(row.id); await quiet(); + }); + + it("keeps a visible first linked issue eligible when a later link is hidden", async () => { + const row = await create(old()), one = await issue(), two = await issue(); + const [first, second] = [one, two].sort((a, b) => a.id.localeCompare(b.id)); + await db.update(issues).set({ hiddenAt: new Date() }).where(eq(issues.id, second.id)); + await issueApprovalService(db).linkManyForApproval(row.id, [second.id, first.id]); + expect((await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true })).items + .some(item => item.subject.id === row.id)).toBe(true); + await start(); await archived(row.id); await quiet(); + }); + + it("joins an admitted sweep on shutdown and skips subsequent deadline queries", async () => { + const row = await create(old()); + const service = approvalRetentionService(db), original = service.sweep; + let release!: () => void, admitted!: () => void; + const barrier = new Promise(resolve => { release = resolve; }); + const admission = new Promise(resolve => { admitted = resolve; }); + vi.spyOn(service, "sweep").mockImplementation(async (now, signal) => { + admitted(); await barrier; return original(now, signal); + }); + const next = vi.spyOn(service, "next"); + const starting = start(service); await admission; + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(100); expect(stopped).toBe(false); + release(); await starting; await stopping; coordinator = undefined; + expect(next).not.toHaveBeenCalled(); expect(await state(row.id)).toBeNull(); + await start(); await archived(row.id); await quiet(); + }); + + + it("uses the shared feed eligibility for a waiting Slack conversation with a pending approval", async () => { + const [agent] = await db.insert(agents).values({ companyId, name: "Origin", role: "engineer", status: "active", adapterType: "codex_local" }).returning(); + const [app] = await db.insert(toolApplications).values({ companyId, name: "Slack", type: "chat" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app.id, name: "Slack", uid: randomUUID(), transport: "chat_sdk", connectionPurpose: "channel" }).returning(); + const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId: connection.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agent.id, status: "active" }).returning(); + const target = await issue(); + await db.update(issues).set({ status: "in_review", assigneeAgentId: agent.id }).where(eq(issues.id, target.id)); + await db.insert(chatConversations).values({ companyId, endpointId: endpoint.id, issueId: target.id, externalConversationId: randomUUID(), externalLabel: "Waiting", state: "waiting" }); + const row = await approvalRecords(db).create(companyId, { type: "request_board_approval", payload: {}, requestedByAgentId: agent.id, updatedAt: old() }); + await issueApprovalService(db).link(target.id, row.id); + const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true }); + expect(feed.items.some(item => item.subject.id === row.id)).toBe(true); + await start(); await archived(row.id); await quiet(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toMatchObject([ + { sourceId: row.id, originIssueId: target.id, originAgentId: agent.id, status: "pending" }, + ]); + }); + + + it("advances past a failed page with full database timestamp precision", async () => { + await db.execute(sql`insert into approvals(company_id, type, payload, updated_at) + select ${companyId}, 'request_board_approval', jsonb_build_object('retentionTestFailure', n <= 100), + date_trunc('milliseconds', now() - interval '91 days') + interval '456 microseconds' + + case when n <= 100 then interval '0 seconds' else interval '1 second' end + from generate_series(1, 101) n`); + await failArchives(); + const service = approvalRetentionService(db); + await expect(service.sweep()).rejects.toThrow("Approval retention failed"); + expect(await service.sweep()).toEqual({ archived: 1 }); + await db.execute(sql`drop trigger approval_retention_fail on decision_retention`); + expect(await service.sweep()).toEqual({ archived: 100 }); + expect(await service.sweep()).toEqual({ archived: 0 }); + expect(await service.next()).toBeNull(); + }); + +}); diff --git a/server/src/__tests__/approvals-service.test.ts b/server/src/__tests__/approvals-service.test.ts index 31755c1f65c..4049b3b0fe0 100644 --- a/server/src/__tests__/approvals-service.test.ts +++ b/server/src/__tests__/approvals-service.test.ts @@ -60,14 +60,12 @@ function createDbStub(selectResults: ApprovalRecord[][], updateResults: Approval const set = vi.fn(() => ({ where: updateWhere })); const update = vi.fn(() => ({ set })); - return { - db: { select, update, transaction: vi.fn(async (callback: (db: unknown) => Promise) => { - inTransaction = true; - try { return await callback({ select, update }); } finally { inTransaction = false; } - }) }, - selectWhere, - returning, - }; + const db = { select, update, transaction: vi.fn(async (callback: (db: unknown) => Promise) => { + const outerTransaction = inTransaction; + inTransaction = true; + try { return await callback(db); } finally { inTransaction = outerTransaction; } + }) }; + return { db, selectWhere, returning }; } describe("approvalService resolution idempotency", () => { diff --git a/server/src/__tests__/attention-service.test.ts b/server/src/__tests__/attention-service.test.ts index bf4879957cd..cf2001e426b 100644 --- a/server/src/__tests__/attention-service.test.ts +++ b/server/src/__tests__/attention-service.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createAgentLifecycle } from "../services/agent-lifecycle.js"; import { randomUUID } from "node:crypto"; import express from "express"; @@ -48,7 +49,8 @@ import { listAttentionExhaustedRuns } from "../services/attention-exhausted-runs import { agentService } from "../services/agents.js"; import { ROUTABLE_BLOCKED_ROLLOUT_AT } from "../services/routable-blocked.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ATTENTION_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -62,8 +64,8 @@ describeEmbeddedPostgres("attention service", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-attention-service-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-attention-service-"); + db = createDb(externalUrl ?? tempDb!.connectionString); }, 30_000); afterEach(async () => { @@ -99,6 +101,7 @@ describeEmbeddedPostgres("attention service", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/budgets-service.test.ts b/server/src/__tests__/budgets-service.test.ts index 1dbf0631a90..0ea5a94dfc6 100644 --- a/server/src/__tests__/budgets-service.test.ts +++ b/server/src/__tests__/budgets-service.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -26,7 +27,8 @@ vi.mock("../services/activity-log.js", () => ({ // Budget behavior is exercised against PostgreSQL below and in // cost-accounting-reliability.test.ts, including transaction and retry paths. -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_BUDGETS_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("budgetService release gate enforcement", () => { @@ -34,8 +36,9 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-budgets-service-"); - db = createDb(tempDb.connectionString); + await import("../services/agent-lifecycle.js"); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-budgets-service-"); + db = createDb(externalUrl ?? tempDb!.connectionString); }, 20_000); afterEach(async () => { @@ -50,6 +53,7 @@ describeEmbeddedPostgres("budgetService release gate enforcement", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 12f46927634..e1d793142a1 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,4 @@ +import { approvalRetentionService } from "../services/approval-retention.js"; import { createSecretProposalsService } from "../services/secret-proposals.js"; import { logger } from "../middleware/logger.js"; import { getStartupRecoveryState } from "../startup-recovery-state.js"; @@ -330,6 +331,11 @@ vi.mock("../realtime/live-events-ws.js", () => ({ setupLiveEventsWebSocketServer: vi.fn(), })); +vi.mock("../services/approval-retention.js", async importOriginal => ({ + ...await importOriginal(), + approvalRetentionService: vi.fn(() => ({ sweep: vi.fn(async () => ({ archived: 0 })), next: vi.fn(async () => null) })), +})); + vi.mock("../services/index.js", () => ({ backfillLegacyToolOAuthTokens: vi.fn(async () => ({ scannedConnections: 0, @@ -530,6 +536,18 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers approval retention only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.approvalRetention); + expect(Boolean(registration)).toBe(enabled); + if (enabled) { + const service = vi.mocked(approvalRetentionService).mock.results.at(-1)!.value; + await registration![1].run(new AbortController().signal); + expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled(); + } + }); + it.each([false, true])("registers archive delivery separately from retention sweeps (scheduler %s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 30000 })); let tick: (() => void) | undefined; diff --git a/server/src/index.ts b/server/src/index.ts index 4ea740f7113..d74ea2b848f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { approvalRetentionService, registerApprovalRetentionWork } from "./services/approval-retention.js"; import { registerDecisionTargetWork } from "./services/decision-target-work.js"; import { registerDecisionEffectRecoveryWork } from "./services/decision-effect-recovery-work.js"; import { registerDecisionExpiryWork } from "./services/decision-expiry-work.js"; @@ -1811,6 +1812,7 @@ async function startServerWithDatabaseTeardown( // per company per tick. const page = await attentionService(db as any).list(company.id, { includeDismissed: true, + excludeSources: ["approval"], all: true, allowUnscopedAll: true, }); @@ -1858,6 +1860,8 @@ async function startServerWithDatabaseTeardown( await step("environment_lease_cleanup", startEnvironmentCleanup); await step("archive_notification_worker", () => registerDecisionArchiveDeliveryWork(deliveryWork, retentionExecutor, () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); + await step("approval_retention", () => registerApprovalRetentionWork(deliveryWork, approvalRetentionService(db as any), + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready); await step("retention_sweep", runRetentionSweep); }; diff --git a/server/src/lib/approval-records.ts b/server/src/lib/approval-records.ts index 957dfac03ec..7bd14df166a 100644 --- a/server/src/lib/approval-records.ts +++ b/server/src/lib/approval-records.ts @@ -1,3 +1,5 @@ +import { notifyToolReviewChange } from "../services/tool-review-work-notifications.js"; +import { writeApprovalChange } from "../services/retention-work-notifications.js"; import { and, eq, inArray } from "drizzle-orm"; import { approvals, type Db } from "@paperclipai/db"; import { notFound, unprocessable } from "../errors.js"; @@ -37,8 +39,9 @@ export function approvalResolutionRecords(db: Db) { } const now = new Date(); - const updated = await database - .update(approvals) + const updated = await writeApprovalChange(database, async tx => { + await notifyToolReviewChange(tx, existing.companyId); + return tx.update(approvals) .set({ status: targetStatus, decidedByUserId, @@ -49,6 +52,7 @@ export function approvalResolutionRecords(db: Db) { .where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses))) .returning() .then((rows) => rows[0] ?? null); + }); if (updated) { return { approval: updated, applied: true }; diff --git a/server/src/services/approval-retention.ts b/server/src/services/approval-retention.ts new file mode 100644 index 00000000000..855c81e413d --- /dev/null +++ b/server/src/services/approval-retention.ts @@ -0,0 +1,67 @@ +import { approvalExecutionIssueCondition } from "./approval-visibility.js"; +import { and, asc, eq, isNull, lte, or, sql, type SQL } from "drizzle-orm"; +import { approvals, companies, decisionRetention, type Db } from "@paperclipai/db"; +import { decisionRetentionService, DEFAULT_DECISION_ARCHIVE_DAYS } from "./decision-retention.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; + +const archiveDelay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; + +export function approvalRetentionService(db: Db) { + let cursor: { at: string; id: string } | undefined; + function candidates(database: Db, condition?: SQL) { + return database.select({ id: approvals.id, companyId: approvals.companyId, at: approvals.updatedAt, cursorAt: sql`${approvals.updatedAt}::text` }).from(approvals) + .innerJoin(companies, eq(companies.id, approvals.companyId)) + .leftJoin(decisionRetention, and(eq(decisionRetention.companyId, approvals.companyId), + eq(decisionRetention.sourceKind, "approval"), sql`${decisionRetention.sourceId} = ${approvals.id}::text`)) + .where(and(sql`${approvals.status} = 'pending'`, eq(companies.status, "active"), approvalExecutionIssueCondition(), + or(isNull(decisionRetention.id), eq(decisionRetention.keep, false)), isNull(decisionRetention.archivedAt), condition)) + .orderBy(asc(approvals.updatedAt), asc(approvals.id)); + } + async function sweep(now = new Date(), signal?: AbortSignal) { + const cutoff = new Date(now.getTime() - archiveDelay); + const rows = await candidates(db, and(lte(approvals.updatedAt, cutoff), cursor + ? sql`(${approvals.updatedAt}, ${approvals.id}) > (${cursor.at}::timestamptz, ${cursor.id}::uuid)` : undefined)).limit(100); + let archived = 0; + const errors: unknown[] = []; + for (const candidate of rows) { + if (signal?.aborted) break; + try { + archived += await db.transaction(async tx => { + const database = tx as unknown as Db; + const [current] = await candidates(database, and(eq(approvals.id, candidate.id), lte(approvals.updatedAt, cutoff))) + .for("update", { of: approvals }); + if (!current) return 0; + const retention = decisionRetentionService(database); + const source = { sourceKind: "approval" as const, subject: { id: current.id }, activityAt: current.at.toISOString() }; + const states = await retention.syncItems(current.companyId, [source]); + const state = states.get(`approval:${current.id}`)!; + return retention.autoArchive({ companyId: current.companyId, now, items: [{ ...source, + keep: state.keep, archivedAt: state.archivedAt?.toISOString() ?? null, retentionVersion: state.version, + }] }); + }); + } catch (error) { errors.push(error); } + cursor = { at: candidate.cursorAt, id: candidate.id }; + } + if (rows.length < 100) cursor = undefined; + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Approval retention failed"); + return { archived }; + } + async function next() { + const [row] = await candidates(db).limit(1); + return row ? row.at.getTime() + archiveDelay : null; + } + return { sweep, next }; +} + +export function registerApprovalRetentionWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.approvalRetention, { + run: signal => service.sweep(new Date(), signal), next: service.next, canRun, retryMs: 30_000, notificationCoalesceMs: 30_000, + }); +} diff --git a/server/src/services/approval-visibility.ts b/server/src/services/approval-visibility.ts new file mode 100644 index 00000000000..3ee21cf174c --- /dev/null +++ b/server/src/services/approval-visibility.ts @@ -0,0 +1,18 @@ +import { sql, type SQL } from "drizzle-orm"; +import { approvals, issues } from "@paperclipai/db"; +import { executionIssueCondition } from "./issue-visibility.js"; + +export function firstApprovalIssueId(): SQL { + return sql`(select first_link.issue_id from issue_approvals first_link + where first_link.company_id = ${approvals.companyId} and first_link.approval_id = ${approvals.id} + order by first_link.issue_id limit 1)`; +} + +export function approvalExecutionIssueCondition(): SQL { + return sql`not exists ( + select 1 from issue_approvals linked where linked.company_id = ${approvals.companyId} + and linked.approval_id = ${approvals.id} and linked.issue_id = ${firstApprovalIssueId()} + and not exists (select 1 from ${issues} where ${issues.id} = linked.issue_id + and ${issues.companyId} = ${approvals.companyId} and ${executionIssueCondition()}) + )`; +} diff --git a/server/src/services/approvals.ts b/server/src/services/approvals.ts index 4f4149f773d..1095766c3ef 100644 --- a/server/src/services/approvals.ts +++ b/server/src/services/approvals.ts @@ -1,4 +1,5 @@ -import { writeToolReviewChange } from "./tool-review-work-notifications.js"; +import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; +import { writeApprovalChange } from "./retention-work-notifications.js"; import { approvalResolutionRecords, resolvableApprovalStatuses as resolvableStatuses } from "../lib/approval-records.js"; import { notifyHireApproved } from "./hire-hook.js"; @@ -13,15 +14,7 @@ import { instanceSettingsService } from "./instance-settings.js"; export function approvalRecords(db: Db) { const instanceSettings = instanceSettingsService(db); - const { getExistingApproval, resolveApproval: resolveApprovalRecord } = approvalResolutionRecords(db); - - // Tool reviews waiting for a formal decision stay parked until this producer - // commits. The shared record owner retains approval validation and CAS rules. - async function resolveApproval(id: string, status: "approved" | "rejected", userId: string, - note: string | null | undefined, database: Db = db) { - const existing = await getExistingApproval(id, database); - return writeToolReviewChange(database, existing.companyId, tx => resolveApprovalRecord(id, status, userId, note, tx)); - } + const { getExistingApproval, resolveApproval } = approvalResolutionRecords(db); function redactApprovalComment(comment: T, censorUsernameInLogs: boolean): T { return { @@ -62,11 +55,11 @@ export function approvalRecords(db: Db) { }, create: (companyId: string, data: Omit) => - db + writeApprovalChange(db, tx => tx .insert(approvals) .values({ ...data, companyId }) .returning() - .then((rows) => rows[0]), + .then((rows) => rows[0])), // Cancel an open (pending/revision_requested) approval without a board // decision — e.g. when its paired agent is terminated during duplicate @@ -75,8 +68,9 @@ export function approvalRecords(db: Db) { const now = new Date(); const [existing] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, id)); if (!existing) return null; - const updated = await writeToolReviewChange(db, existing.companyId, tx => tx - .update(approvals) + const updated = await writeApprovalChange(db, async tx => { + await notifyToolReviewChange(tx, existing.companyId); + return tx.update(approvals) .set({ status: "cancelled", decisionNote: reason ?? null, @@ -85,7 +79,8 @@ export function approvalRecords(db: Db) { }) .where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses))) .returning() - .then((rows) => rows[0] ?? null)); + .then((rows) => rows[0] ?? null); + }); return updated; }, @@ -96,7 +91,7 @@ export function approvalRecords(db: Db) { } const now = new Date(); - return db + return writeApprovalChange(db, tx => tx .update(approvals) .set({ status: "revision_requested", @@ -107,7 +102,7 @@ export function approvalRecords(db: Db) { }) .where(eq(approvals.id, id)) .returning() - .then((rows) => rows[0]); + .then((rows) => rows[0])); }, resubmit: async (id: string, payload?: Record) => { @@ -117,7 +112,7 @@ export function approvalRecords(db: Db) { } const now = new Date(); - return db + return writeApprovalChange(db, tx => tx .update(approvals) .set({ status: "pending", @@ -129,7 +124,7 @@ export function approvalRecords(db: Db) { }) .where(eq(approvals.id, id)) .returning() - .then((rows) => rows[0]); + .then((rows) => rows[0])); }, listComments: async (approvalId: string) => { diff --git a/server/src/services/attention.ts b/server/src/services/attention.ts index 90ed24f135e..1e94cd26ec3 100644 --- a/server/src/services/attention.ts +++ b/server/src/services/attention.ts @@ -1,3 +1,4 @@ +import { approvalExecutionIssueCondition, firstApprovalIssueId } from "./approval-visibility.js"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, notInArray, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -161,6 +162,7 @@ type AttentionListOptions = AttentionFeedQuery & { actor?: AuthorizationActor; /** Internal-only escape hatch for callers that need one stable, unpaginated feed snapshot. */ allowUnscopedAll?: boolean; + excludeSources?: readonly AttentionSourceKind[]; }; type AttentionServiceOptions = { @@ -1185,83 +1187,72 @@ export function attentionService(db: Db, serviceOptions: AttentionServiceOptions collected.push({ ...item, dismissal }); }; - const pendingApprovals = await db - .select({ - id: approvals.id, - type: approvals.type, - status: approvals.status, - requestedByAgentId: approvals.requestedByAgentId, - requestedByUserId: approvals.requestedByUserId, - payload: approvals.payload, - createdAt: approvals.createdAt, - updatedAt: approvals.updatedAt, - }) - .from(approvals) - .where(and(eq(approvals.companyId, companyId), eq(approvals.status, "pending"))) - .orderBy(desc(approvals.updatedAt), desc(approvals.id)); + if (!options.excludeSources?.includes("approval")) { + const pendingApprovals = await db + .select({ + id: approvals.id, + linkedIssueId: firstApprovalIssueId(), + type: approvals.type, + status: approvals.status, + requestedByAgentId: approvals.requestedByAgentId, + requestedByUserId: approvals.requestedByUserId, + payload: approvals.payload, + createdAt: approvals.createdAt, + updatedAt: approvals.updatedAt, + }) + .from(approvals) + .where(and(eq(approvals.companyId, companyId), eq(approvals.status, "pending"), approvalExecutionIssueCondition())) + .orderBy(desc(approvals.updatedAt), desc(approvals.id)); - const pendingApprovalIds = pendingApprovals.map((approval) => approval.id); - const approvalIssueRows = pendingApprovalIds.length > 0 - ? await db - .select({ approvalId: issueApprovals.approvalId, issueId: issueApprovals.issueId }) - .from(issueApprovals) - .where(and( - eq(issueApprovals.companyId, companyId), - inArray(issueApprovals.approvalId, pendingApprovalIds), - )) - .orderBy(asc(issueApprovals.approvalId), asc(issueApprovals.issueId)) - : []; - const approvalIssueMap = new Map(); - for (const row of approvalIssueRows) { - if (!approvalIssueMap.has(row.approvalId)) approvalIssueMap.set(row.approvalId, row.issueId); - } - const readableApprovalIssues = await issueSummaryMap( - db, - companyId, - approvalIssueRows.map((row) => row.issueId), - readCondition, - ); - - for (const approval of pendingApprovals) { - const linkedIssueId = approvalIssueMap.get(approval.id) ?? null; - if (linkedIssueId && !readableApprovalIssues.has(linkedIssueId)) continue; - const dedupKey = `approval:${approval.id}`; - const title = approvalTitle(approval.type, approval.payload); - add(createItem({ + const readableApprovalIssues = await issueSummaryMap( + db, companyId, - sourceKind: "approval", - subject: { - kind: "approval", - id: approval.id, + pendingApprovals.map((row) => row.linkedIssueId), + readCondition, + ); + + for (const approval of pendingApprovals) { + const linkedIssueId = approval.linkedIssueId ?? null; + if (linkedIssueId && !readableApprovalIssues.has(linkedIssueId)) continue; + const dedupKey = `approval:${approval.id}`; + const title = approvalTitle(approval.type, approval.payload); + add(createItem({ companyId, - title, - identifier: null, - status: approval.status, - href: `/${prefix}/approvals/${approval.id}`, - metadata: { - type: approval.type, - requestedByAgentId: approval.requestedByAgentId, - requestedByUserId: approval.requestedByUserId, - issueId: approvalIssueMap.get(approval.id) ?? null, + sourceKind: "approval", + subject: { + kind: "approval", + id: approval.id, + companyId, + title, + identifier: null, + status: approval.status, + href: `/${prefix}/approvals/${approval.id}`, + metadata: { + type: approval.type, + requestedByAgentId: approval.requestedByAgentId, + requestedByUserId: approval.requestedByUserId, + issueId: approval.linkedIssueId ?? null, + }, }, - }, - whyNow: "Approval is pending a board decision.", - decisionVerbs: decisionVerbs( - { id: "approve", label: "Approve", description: "Approve the request." }, - { id: "reject", label: "Reject", description: "Reject the request." }, - { id: "request_revision", label: "Request revision", description: "Send the request back for changes." }, - ), - inlineResolvable: approval.type !== "request_board_approval", - entryRule: "approvals.status = 'pending'", - exitRule: "Approval leaves pending status.", - dedupKey, - severity: "medium", - activityAt: toIso(approval.updatedAt), - createdAt: toIso(approval.createdAt), - updatedAt: toIso(approval.updatedAt), - relatedIssue: null, - detail: approvalDetail(approval.type, approval.payload), - })); + whyNow: "Approval is pending a board decision.", + decisionVerbs: decisionVerbs( + { id: "approve", label: "Approve", description: "Approve the request." }, + { id: "reject", label: "Reject", description: "Reject the request." }, + { id: "request_revision", label: "Request revision", description: "Send the request back for changes." }, + ), + inlineResolvable: approval.type !== "request_board_approval", + entryRule: "approvals.status = 'pending'", + exitRule: "Approval leaves pending status.", + dedupKey, + severity: "medium", + activityAt: toIso(approval.updatedAt), + createdAt: toIso(approval.createdAt), + updatedAt: toIso(approval.updatedAt), + relatedIssue: null, + detail: approvalDetail(approval.type, approval.payload), + })); + } + } const interactionRows = await db diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index bf02e510ca0..84e4ca1e6dc 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "./retention-work-notifications.js"; import { addCents, compareCents, normalizeCents } from "@paperclipai/shared"; import { and, desc, eq, inArray, sql, or, isNull } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -258,6 +259,7 @@ async function markApprovalStatus( decidedByUserId: string, ) { if (!approvalId) return; + await notifyRetentionWork(db, "approval"); await db .update(approvals) .set({ @@ -295,6 +297,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.paused) return; + await notifyRetentionWork(db, "approval"); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); await db @@ -351,6 +354,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica await notifyDeliveryWork(db, DELIVERY_QUEUES.skillSourceSync); const scope = await resolveScopeRecord(db, "company", policy.scopeId); if (scope.pauseReason !== "budget") return; + await notifyRetentionWork(db, "approval"); await notifyDeliveryWork(db, DELIVERY_QUEUES.agentLifecycle, policy.companyId); await db .update(companies) @@ -453,6 +457,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica windowEnd: end, }); + if (thresholdType === "hard") await notifyRetentionWork(db, "approval"); const approval = thresholdType === "hard" ? await db .insert(approvals) diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index fa29b357f3b..f6fc7ed5de2 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { isRunnerApiResponseAsset } from "./asset-content-access.js"; import { deleteCompany } from "./company-deletion.js"; @@ -330,6 +331,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyRetentionWork(tx, "approval"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); } const updated = await tx @@ -481,6 +483,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); + await notifyRetentionWork(tx, "approval"); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); await tx .update(companies) diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index ab549ecbfb8..c3b3aaf3e13 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "./retention-work-notifications.js"; import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { agentObservabilityService } from "./agent-observability.js"; @@ -96,6 +97,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(agentRuntimeState).where(eq(agentRuntimeState.companyId, id)); await tx.delete(issueComments).where(eq(issueComments.companyId, id)); await tx.delete(approvalComments).where(eq(approvalComments.companyId, id)); + await notifyRetentionWork(tx, "approval"); await tx.delete(approvals).where(eq(approvals.companyId, id)); await tx.delete(companySecrets).where(eq(companySecrets.companyId, id)); await tx.delete(joinRequests).where(eq(joinRequests.companyId, id)); diff --git a/server/src/services/decision-retention.ts b/server/src/services/decision-retention.ts index b65e6e3bb77..7b7a993cce4 100644 --- a/server/src/services/decision-retention.ts +++ b/server/src/services/decision-retention.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "./retention-work-notifications.js"; import { decisionArchiveDeliveryService, type ArchiveNotificationBatch } from "./decision-archive-delivery.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { createHash } from "node:crypto"; @@ -32,6 +33,9 @@ export const DEFAULT_DECISION_SHELF_DAYS = 30; export const DEFAULT_DECISION_ARCHIVE_DAYS = 90; const DAY_MS = 86_400_000; +type RetentionSource = Pick & { subject: Pick }; +type ArchiveSource = RetentionSource & Pick; + export type DecisionRetentionState = typeof decisionRetention.$inferSelect; export type { ArchiveNotificationBatch } from "./decision-archive-delivery.js"; @@ -128,45 +132,49 @@ export function decisionRetentionService( db: Db, options: { notifyOriginAgent?: (batch: ArchiveNotificationBatch) => Promise } = {}, ) { - async function syncItems(companyId: string, items: readonly AttentionItem[]) { + async function syncItems(companyId: string, items: readonly RetentionSource[]) { if (items.length === 0) return new Map(); - const unique = [...new Map(items.map((item) => [itemSourceKey(item), item])).values()]; - await db.insert(decisionRetention).values(unique.map((item) => ({ - companyId, - sourceKind: item.sourceKind, - sourceId: item.subject.id, - sourceActivityAt: new Date(item.activityAt), - }))).onConflictDoNothing({ - target: [decisionRetention.companyId, decisionRetention.sourceKind, decisionRetention.sourceId], + const unique = [...new Map(items.map(item => [itemSourceKey(item), item])).values()]; + return db.transaction(async tx => { + const select = () => tx.select().from(decisionRetention).where(and( + eq(decisionRetention.companyId, companyId), + inArray(decisionRetention.sourceId, unique.map(item => item.subject.id)), + )); + let rows = await select(); + const existing = new Set(rows.map(row => sourceKey(row.sourceKind, row.sourceId))); + const missing = unique.filter(item => !existing.has(itemSourceKey(item))); + if (missing.length) { + for (const kind of new Set(missing.map(item => item.sourceKind))) await notifyRetentionWork(tx, kind); + await tx.insert(decisionRetention).values(missing.map(item => ({ + companyId, sourceKind: item.sourceKind, sourceId: item.subject.id, sourceActivityAt: new Date(item.activityAt), + }))).onConflictDoNothing({ target: [decisionRetention.companyId, decisionRetention.sourceKind, decisionRetention.sourceId] }); + rows = await select(); + } + const byKey = new Map(rows.map(row => [sourceKey(row.sourceKind, row.sourceId), row])); + for (const item of unique) { + const key = itemSourceKey(item), row = byKey.get(key), activityAt = new Date(item.activityAt); + if (!row || row.sourceActivityAt.getTime() === activityAt.getTime()) continue; + await notifyRetentionWork(tx, item.sourceKind); + const [updated] = await tx.update(decisionRetention).set({ + sourceActivityAt: activityAt, version: sql`${decisionRetention.version} + 1`, updatedAt: new Date(), + }).where(and(eq(decisionRetention.id, row.id), ne(decisionRetention.sourceActivityAt, activityAt))).returning(); + if (updated) byKey.set(key, updated); + } + return byKey; }); - let rows = await db.select().from(decisionRetention).where(and( - eq(decisionRetention.companyId, companyId), - inArray(decisionRetention.sourceId, unique.map((item) => item.subject.id)), - )); - const byKey = new Map(rows.map((row) => [sourceKey(row.sourceKind, row.sourceId), row])); - for (const item of unique) { - const key = itemSourceKey(item); - const row = byKey.get(key); - const activityAt = new Date(item.activityAt); - if (!row || row.sourceActivityAt.getTime() === activityAt.getTime()) continue; - const updated = await db.update(decisionRetention).set({ - sourceActivityAt: activityAt, - version: sql`${decisionRetention.version} + 1`, - updatedAt: new Date(), - }).where(and( - eq(decisionRetention.id, row.id), - ne(decisionRetention.sourceActivityAt, activityAt), - )).returning().then((values) => values[0] ?? null); - if (updated) byKey.set(key, updated); - } - rows = [...byKey.values()]; - return new Map(rows.map((row) => [sourceKey(row.sourceKind, row.sourceId), row])); } - function itemSourceKey(item: AttentionItem) { + function itemSourceKey(item: RetentionSource) { return sourceKey(item.sourceKind, item.subject.id); } + async function writeRetention(kind: AttentionSourceKind, write: (tx: Db) => Promise) { + return db.transaction(async tx => { + await notifyRetentionWork(tx, kind); + return write(tx as unknown as Db); + }); + } + async function getState(companyId: string, sourceKind: AttentionSourceKind, sourceId: string) { return db.select().from(decisionRetention).where(and( eq(decisionRetention.companyId, companyId), @@ -175,6 +183,16 @@ export function decisionRetentionService( )).then((rows) => rows[0] ?? null); } + async function materializeApprovalState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string }) { + if (input.sourceKind !== "approval") return; + const [source] = await database.select({ updatedAt: approvals.updatedAt }).from(approvals) + .where(and(eq(approvals.companyId, input.companyId), eq(approvals.id, input.sourceId))).for("update"); + if (!source) throw notFound("Attention source not found"); + await decisionRetentionService(database).syncItems(input.companyId, [{ + sourceKind: "approval", subject: { id: input.sourceId }, activityAt: source.updatedAt.toISOString(), + }]); + } + async function setKeep(input: { companyId: string; sourceKind: AttentionSourceKind; @@ -186,15 +204,18 @@ export function decisionRetentionService( if (!(await canReadDecisionSource(db, input.authActor, input.companyId, input.sourceKind, input.sourceId))) { throw notFound("Attention source not found"); } - const updated = await db.update(decisionRetention).set({ - keep: input.keep, - version: sql`${decisionRetention.version} + 1`, - updatedAt: new Date(), - }).where(and( - eq(decisionRetention.companyId, input.companyId), - eq(decisionRetention.sourceKind, input.sourceKind), - eq(decisionRetention.sourceId, input.sourceId), - )).returning().then((rows) => rows[0] ?? null); + const updated = await writeRetention(input.sourceKind, async tx => { + await materializeApprovalState(tx, input); + return tx.update(decisionRetention).set({ + keep: input.keep, + version: sql`${decisionRetention.version} + 1`, + updatedAt: new Date(), + }).where(and( + eq(decisionRetention.companyId, input.companyId), + eq(decisionRetention.sourceKind, input.sourceKind), + eq(decisionRetention.sourceId, input.sourceId), + )).returning().then((rows) => rows[0] ?? null); + }); if (!updated) throw notFound("Attention source not found"); await logActivity(db, { companyId: input.companyId, @@ -224,28 +245,31 @@ export function decisionRetentionService( throw notFound("Attention source not found"); } const now = new Date(); - const updated = await db.update(decisionRetention).set(input.archived ? { - archivedAt: now, - archivedReason: "manual", - ...archiveActorColumns(input.actor), - archiveVersion: sql`${decisionRetention.archiveVersion} + 1`, - version: sql`${decisionRetention.version} + 1`, - updatedAt: now, - } : { - archivedAt: null, - archivedReason: null, - archivedByType: null, - archivedByAgentId: null, - archivedByUserId: null, - archivedByRunId: null, - version: sql`${decisionRetention.version} + 1`, - updatedAt: now, - }).where(and( - eq(decisionRetention.companyId, input.companyId), - eq(decisionRetention.sourceKind, input.sourceKind), - eq(decisionRetention.sourceId, input.sourceId), - input.archived ? isNull(decisionRetention.archivedAt) : isNotNull(decisionRetention.archivedAt), - )).returning().then((rows) => rows[0] ?? null); + const updated = await writeRetention(input.sourceKind, async tx => { + await materializeApprovalState(tx, input); + return tx.update(decisionRetention).set(input.archived ? { + archivedAt: now, + archivedReason: "manual", + ...archiveActorColumns(input.actor), + archiveVersion: sql`${decisionRetention.archiveVersion} + 1`, + version: sql`${decisionRetention.version} + 1`, + updatedAt: now, + } : { + archivedAt: null, + archivedReason: null, + archivedByType: null, + archivedByAgentId: null, + archivedByUserId: null, + archivedByRunId: null, + version: sql`${decisionRetention.version} + 1`, + updatedAt: now, + }).where(and( + eq(decisionRetention.companyId, input.companyId), + eq(decisionRetention.sourceKind, input.sourceKind), + eq(decisionRetention.sourceId, input.sourceId), + input.archived ? isNull(decisionRetention.archivedAt) : isNotNull(decisionRetention.archivedAt), + )).returning().then((rows) => rows[0] ?? null); + }); if (!updated) { const current = await getState(input.companyId, input.sourceKind, input.sourceId); if (!current) throw notFound("Attention source not found"); @@ -298,6 +322,7 @@ export function decisionRetentionService( } const now = new Date(); for (const entry of input.manifest) { + await notifyRetentionWork(txDb, entry.sourceKind as AttentionSourceKind); const archived = await txDb.update(decisionRetention).set({ archivedAt: now, archivedReason: "accepted_proposal", @@ -333,7 +358,7 @@ export function decisionRetentionService( }); } - async function autoArchive(input: { companyId: string; items: AttentionItem[]; now?: Date }) { + async function autoArchive(input: { companyId: string; items: readonly ArchiveSource[]; now?: Date }) { const now = input.now ?? new Date(); const cutoff = new Date(now.getTime() - DEFAULT_DECISION_ARCHIVE_DAYS * DAY_MS); const candidates = input.items.filter((item) => new Date(item.activityAt) <= cutoff && !item.keep && !item.archivedAt); @@ -341,6 +366,7 @@ export function decisionRetentionService( await db.transaction(async (tx) => { const txDb = tx as unknown as Db; for (const item of candidates) { + await notifyRetentionWork(txDb, item.sourceKind); const updated = await txDb.update(decisionRetention).set({ archivedAt: now, archivedReason: "idle_ttl", diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 59389821c89..55f7943d441 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + approvalRetention: "approval-retention", decisionTargets: "decision-targets", decisionEffectRecovery: "decision-effect-recovery", decisionExpiry: "decision-expiry", diff --git a/server/src/services/issue-approvals.ts b/server/src/services/issue-approvals.ts index 9d2d0a371ac..b00133898b0 100644 --- a/server/src/services/issue-approvals.ts +++ b/server/src/services/issue-approvals.ts @@ -1,3 +1,4 @@ +import { writeApprovalChange } from "./retention-work-notifications.js"; import { and, desc, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { approvals, issueApprovals, issues } from "@paperclipai/db"; @@ -107,7 +108,7 @@ export function issueApprovalService(db: Db) { link: async (issueId: string, approvalId: string, actor?: LinkActor) => { const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId); - await db + await writeApprovalChange(db, tx => tx .insert(issueApprovals) .values({ companyId: issue.companyId, @@ -116,7 +117,7 @@ export function issueApprovalService(db: Db) { linkedByAgentId: actor?.agentId ?? null, linkedByUserId: actor?.userId ?? null, }) - .onConflictDoNothing(); + .onConflictDoNothing()); return db .select() @@ -126,10 +127,10 @@ export function issueApprovalService(db: Db) { }, unlink: async (issueId: string, approvalId: string) => { - await assertIssueAndApprovalSameCompany(issueId, approvalId); - await db + const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId); + await writeApprovalChange(db, tx => tx .delete(issueApprovals) - .where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId))); + .where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId)))); }, linkManyForApproval: async (approvalId: string, issueIds: string[], actor?: LinkActor) => { @@ -157,7 +158,7 @@ export function issueApprovalService(db: Db) { } } - await db + await writeApprovalChange(db, tx => tx .insert(issueApprovals) .values( uniqueIssueIds.map((issueId) => ({ @@ -168,7 +169,7 @@ export function issueApprovalService(db: Db) { linkedByUserId: actor?.userId ?? null, })), ) - .onConflictDoNothing(); + .onConflictDoNothing()); }, }; } diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 1e77167e85d..53f3cff74a6 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "./retention-work-notifications.js"; import { notifyDecisionTargetWork, notifyDecisionTargetIssueRemoval } from "./decision-targets.js"; import { assertOnboardingFirstTaskCreationAllowed } from "./onboarding-first-task-admission.js"; @@ -11389,6 +11390,9 @@ export function issueService(db: Db) { if (patch.status === "done" || patch.status === "cancelled") { await notifyDecisionTargetWork(tx, `issue:${id}`); } + if (patch.hiddenAt !== undefined || patch.harnessKind !== undefined || patch.conversationAgentId !== undefined) { + await notifyRetentionWork(tx, "approval"); + } const updated = await tx .update(issues) .set(patch) @@ -11796,6 +11800,7 @@ export function issueService(db: Db) { try { await notifyTerminalWorkspaceWork(tx, `issue:${id}`); await requestRestorePointDeletion(tx, { sourceIssueId: id }); + await notifyRetentionWork(tx, "approval"); await notifyDecisionTargetIssueRemoval(tx, id); removedIssue = await tx .delete(issues) diff --git a/server/src/services/retention-work-notifications.ts b/server/src/services/retention-work-notifications.ts new file mode 100644 index 00000000000..3464c624e89 --- /dev/null +++ b/server/src/services/retention-work-notifications.ts @@ -0,0 +1,14 @@ +import type { Db } from "@paperclipai/db"; +import type { AttentionSourceKind } from "@paperclipai/shared"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export async function notifyRetentionWork(transaction: object, kind: AttentionSourceKind) { + if (kind === "approval") await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention); +} + +export function writeApprovalChange(db: Db, write: (transaction: Db) => Promise): Promise { + return db.transaction(async tx => { + await notifyRetentionWork(tx, "approval"); + return write(tx as unknown as Db); + }); +} diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 50c200490b7..372c8d4f45d 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -1,3 +1,4 @@ +import { notifyRetentionWork } from "./retention-work-notifications.js"; import { notifyDecisionTargetIssueRemoval } from "./decision-targets.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { verifyAppWebhook } from "./app-webhook.js"; @@ -1965,6 +1966,7 @@ export function routineService( return updated ?? createdRun; } catch (error) { if (createdIssue) { + await notifyRetentionWork(txDb, "approval"); await notifyDecisionTargetIssueRemoval(txDb, createdIssue.id); await txDb.delete(issues).where(eq(issues.id, createdIssue.id)); } diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 4cacb30e47e..ae81998e947 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,3 +1,4 @@ +import { writeApprovalChange } from "./retention-work-notifications.js"; import { toolReviewProjectionService } from "./tool-review-projection.js"; import { toolReviewInterruptionService } from "./tool-review-interruption.js"; import { bindVerifiedToolApprovalArguments, claimToolTestApproval, releaseToolApproval, persistToolTestOutcome, startToolInvocation, type ToolApprovalClaim } from "./tool-approval-execution.js"; @@ -2649,7 +2650,7 @@ export function createToolGatewayService( let formalApprovalId: string | null = null; if (toolRequiresFormalApproval(input.tool)) { - const [approval] = await db + const [approval] = await writeApprovalChange(db, tx => tx .insert(approvals) .values({ companyId: input.session.companyId, @@ -2672,17 +2673,18 @@ export function createToolGatewayService( argumentsHash: canonicalArgumentsHash, }, }) - .returning(); + .returning()); formalApprovalId = approval.id; - await db + const approvalIssueId = input.session.issueId; + await writeApprovalChange(db, tx => tx .insert(issueApprovals) .values({ companyId: input.session.companyId, - issueId: input.session.issueId, + issueId: approvalIssueId, approvalId: approval.id, linkedByAgentId: input.session.agentId, }) - .onConflictDoNothing(); + .onConflictDoNothing()); } const interaction = await interactions.create(