From 553a004926281e879b5823987fc057e544fa686a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:23:24 -0500 Subject: [PATCH] fix(server): schedule approval retention from durable deadlines (#17497) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip manages agent work and should become idle when it owes no background work. > - The attention feed archives old pending approvals under the existing 90-day policy. > - The server rebuilds the feed on each maintenance tick, even when approval state has not changed. > - Some approvals have never been read, so retention rows alone cannot define the work. > - Approval records and committed eligibility changes define the real deadline and wake triggers. > - This PR gives approval retention its own deadline worker and reuses the existing retention and archive-delivery contracts. > - The approval lane stops idle scans while other attention sources remain separate migration work. ## Linked Issues or Issue Description Stacked on #17490. This PR owns approval retention only. Archive notification delivery is already separated in #17475. Other attention-source retention and remaining recurring server duties remain in the stack inventory. The related-PR search found no overlapping approval-retention worker migration. **What existing behavior does this improve?** Automatic archival of old pending approvals from attention. **Current behavior** The shared maintenance tick builds the attention feed for each active company. That query discovers pending approvals and materializes retention records before checking their age. An unchanged or empty approval source is read repeatedly. **Proposed behavior** Reconstruct eligible approvals at startup. Schedule the earliest approval activity timestamp plus 90 days. Wake after committed changes to approvals, links, issue visibility, company activity, or retention controls. Stop scheduling when no eligible approval remains. Archive records through the existing version checks, audit writes, and notification outbox. **Reason and benefit** Idle approval retention should generate no recurring SQL. Unseen approvals still need eventual archival. A stale candidate must not archive an approval that was resubmitted while the worker waited for its source lock. **Breaking changes** None intended. The archive age, eligibility, keep controls, audit, and notification policy remain. Normal attention reads still include approvals. No Cloud sleep blocker is relaxed. ## What Changed - Restrict tool-review signals to their existing terminal transition owners (approve/reject/cancel). Creation/link/revision/resubmit signal retention only. - Coalesce retention notifications under one thirty-second aggregate read budget. Known deadlines still preempt it; no notification or durable deadline means no scan. - Add a bounded approval-retention worker. It uses the shared deadline coordinator and reads pending approvals directly. - Share first-linked-issue ordering and execution visibility through `approval-visibility.ts`, used by both the feed and worker. Preserve actor-specific read filtering and locked archival rechecks. - Materialize unseen approvals on demand inside manual keep/archive/revive transactions. Use the saved activity timestamp and existing versioned state; retain company scoping and rollback on failure. - Add awaited pre-write signals to approval, budget, tool, link, visibility, company, and retention mutation owners. Preserve outer transaction ownership; keep provider calls outside SQL wrappers. - Narrow retention inputs to the source fields the existing service uses. Signal materialization only for inserted or changed state so unchanged reads converge. - Exclude approval collection from the maintenance feed before its query. Other source kinds remain in that sweep. - Add the pending-only approval deadline index with concurrent, idempotent creation. Keep the five newest migration snapshots. - Add real-Postgres lifecycle tests and startup wiring tests. Enable validated external test databases for existing attention and budget suites. - Document producer coverage, quiet conditions, recovery and remaining scope in `doc/idle-approval-retention.md` and the stack ledger. ## Verification Current test revision: `dd844f4e8cbbf1db681a774dc31cd552301bef48`, followed by docs-only `3efda33812`, which received both 5/5 reviews and passing CI. This candidate normally integrates landed Grok `2adcde162e`. All runtime and acceptance-test sources are byte-identical to that tested/reviewed tree; only four independent Runner E2E accounting files and the ledger differ. Migration 0392 and the parent journal prefix are unchanged. Local checks are reused for their unchanged source; fresh final-head CI and reviews are required against master. This remains the approval-retention duty. - **315 cases pass** across eight complete suites: approval retention/service, attention, decision retention, budgets, tool gateway, server startup and scoped deadlines. All 30 real-PostgreSQL lifecycle cases pass. Server production typecheck and build pass. - **Four regressions failed before repair**: unseen young keep/archive/revive returned 404; the manual keep rollback case could not reach its update. They now pass. A failed keep update leaves no retention row or audit; retry succeeds with one audit. - Both linked-issue orders are checked against the feed: hidden first/visible later is excluded; visible first/hidden later remains eligible. The feed and worker use the same owner query, with actor read filtering preserved. - Generated concurrent/idempotent migration `0392_small_frightful_four.sql` adds only the pending approval index. Every historical parent SQL migration and all 390 parent journal entries are unchanged. Five generated snapshots remain (0388–0392). Schema drift, migration safety/numbering and actual fresh apply/replay on five isolated PostgreSQL databases pass. - Existing precision, transaction, drain/resume, failure fairness and shutdown tests remain. Earlier transaction-stub, budget-import, diagnostic-client and CI failures are recorded in the owning audit and stack ledger; later passes do not erase them. - Three additional real regressions failed before the wake repair: ordinary mutations emitted tool-review signals in both resolution cases, and sequential tenant writes caused repeated global reads. Tests now prove terminal approve/reject/cancel signals and one aggregate reconciliation per thirty-second window with a different company's pending approval retained. The aggregate budget is stricter than a separate per-company budget and matches the old default cadence; known deadlines preempt it. - The initial coalescing fixture expected the concrete pending-work hold to clear before reconciliation; it now preserves that hold and measures SQL. The first broad repair passed 313 cases but failed the existing parked-review cancel case. The original cancel producer is restored; all 315 now pass without weakening assertions. At old head `5991f257bb`, security was action-required and Greptile returned service error `8fb1835c-e564-476b-b802-d5be164ea915`. Fresh repaired-head results are required, not assumed. Commands: server Vitest for `approval-retention-work`, `approvals-service`, `attention-service`, `decision-retention-service`, `budgets-service`, `tool-gateway-service`, `server-startup-feedback-export`, `scoped-deadline-work`; server typecheck/build; DB snapshot drift, migration checks and fresh apply/replay. Fresh exact-head Greptile/Ponytail and CI remain merge gates. No full local repository suite or deployed host suspension/cost claim is made. Earlier stack verification below is historical evidence at the explicitly named revisions, not this head's acceptance: Approved delegated Test-tab calls now evaluate caller and target policies in the existing claim transaction. Real HTTP regressions first showed a missing caller charge and shared-limit self-rejection. All 78 connection-management cases and 92 gateway cases now pass on the owning branch, along with server production typechecking. All 78 connection-management cases also pass at combined tip `d25f830db89c1623bff3547bfc68015c1c66e0cf`. The three new cases cover shared charges, separate charges, rollback on denial, duplicate approval, and rejection of a later authorized call. Intermediate counter/identity fixture failures and a reused-database Slack collision are recorded in [the owning audit](https://github.com/paperclipai/paperclip/blob/20557d2c3dd0949c734eaa8095fd78b9b07d7fd5/doc/idle-approved-tool-recovery.md#delegated-test-tab-admission-after-policy-changes). The earlier combined checks below remain evidence for their stated revisions. Current-head CI and review remain required. Earlier published head: `20557d2c3dd0949c734eaa8095fd78b9b07d7fd5`. This remains a single-duty PR stacked on #17490. - Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5` with normal merges. Every branch passed generated-schema checks and its target-branch migration-order gate. Existing SQL is unchanged; the 25 stack migrations follow master’s new migration. - Duty-specific producer coverage, lifecycle tests, and earlier failure records: [2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/76a49810c7ff403ae7720bef0d453747bd844731/doc/plans/2026-10-24-recurring-server-sleep-stack.md). - Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355 startup/cancellation/recovery/dispatch/preservation tests**, **75 connection-management tests**, and production server typechecking pass. These combined checks do not mean every intermediate PR independently ran the full repository suite. - Real master-to-stack database upgrade passes: **379 unchanged prior migration receipts plus 25 new migrations**, with the gateway responsibility fixture preserved. - Owning-branch checks: 84 runner installation/settlement cases plus runner typechecking and generated TypeScript/Rust profile agreement; 253 GitHub review workflow cases. Stop-attribution and held-worker readiness regressions failed before their fixes. - Failure accounting: the initial combined run passed 349 cases and failed six idle assertions in a reused preservation database containing old cancelled runs. Repeating the combined command with a fresh preservation database passed. Earlier invalid project selection, merge import/type errors, and the interrupted full local suite are not counted as passing checks. - **Latest-head CI and review remain required.** Prior CI findings include GitHub idle observation, route teardown, attention cleanup, touch-picker E2E, workspace health, and voice-cleanup timing; local passes do not erase those failures. No full local repository-suite, deployed sleep, fleet cost, or live Copilot qualification claim is made. ## Risks - Signals cover the application database owner and its child pools. Independent SQL writers and other active processes need an explicit wake integration. Startup reconstructs committed durable state. - The source row lock prevents stale approval-activity archival. It does not add cross-table serializability for concurrent link, visibility, or company-policy changes. - Newly changed eligibility can wait up to the thirty-second notification window, matching the former default maintenance cadence. Saved deadlines still preempt; no new idle polling is added. - Failed records retain recovery debt. Bounded cursor progress lets later rows proceed without dropping failures. - The shared retention transaction now groups materialization and activity-version updates. Existing attention and retention consumers are tested against PostgreSQL. - Kept, archived, resolved or inactive-company approvals can become quiet, but their supported mutations must retain their wake notifications. - The local scheduler cannot wake stopped compute. This PR neither adds approval deadlines to Cloud's external alarm contract nor claims measured host or database suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/idle-approval-retention.md | 164 +++++++ ...2026-10-24-recurring-server-sleep-stack.md | 224 +++++++++- packages/db/src/client.ts | 1 + .../migrations/0392_small_frightful_four.sql | 1 + ...{0387_snapshot.json => 0392_snapshot.json} | 108 ++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/approvals.ts | 2 + .../__tests__/approval-retention-work.test.ts | 423 ++++++++++++++++++ .../src/__tests__/approvals-service.test.ts | 14 +- .../src/__tests__/attention-service.test.ts | 9 +- server/src/__tests__/budgets-service.test.ts | 10 +- .../server-startup-feedback-export.test.ts | 18 + server/src/index.ts | 4 + server/src/lib/approval-records.ts | 8 +- server/src/services/approval-retention.ts | 67 +++ server/src/services/approval-visibility.ts | 18 + server/src/services/approvals.ts | 33 +- server/src/services/attention.ts | 137 +++--- server/src/services/budgets.ts | 5 + server/src/services/companies.ts | 3 + server/src/services/company-deletion.ts | 2 + server/src/services/decision-retention.ts | 154 ++++--- .../services/delivery-work-notifications.ts | 1 + server/src/services/issue-approvals.ts | 15 +- server/src/services/issues.ts | 5 + .../services/retention-work-notifications.ts | 14 + server/src/services/routines.ts | 2 + server/src/services/tool-gateway.ts | 12 +- 28 files changed, 1273 insertions(+), 188 deletions(-) create mode 100644 doc/idle-approval-retention.md create mode 100644 packages/db/src/migrations/0392_small_frightful_four.sql rename packages/db/src/migrations/meta/{0387_snapshot.json => 0392_snapshot.json} (99%) create mode 100644 server/src/__tests__/approval-retention-work.test.ts create mode 100644 server/src/services/approval-retention.ts create mode 100644 server/src/services/approval-visibility.ts create mode 100644 server/src/services/retention-work-notifications.ts 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(