From d108aeeaaf7c4a3050b3650444884e6c766b46d8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 15:26:19 -0500 Subject: [PATCH] fix(server): schedule pending tool-review expiry from committed changes (#17453) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Governed tools can wait for a human review before execution. > - Pending reviews already save a deadline and must reject approval after expiry. > - The server currently scans for expired reviews on every maintenance tick. > - Empty scans keep the database active without advancing work. > - This pull request schedules pending review expiry from committed changes and saved deadlines. > - Configured idle state and resolved history no longer cause this expiry scan. ## Linked Issues or Issue Description **What happened?** The combined tool review sweep repeatedly queries pending expiry even when no pending request has a deadline. **What did you expect?** Signing or resolving a request should update the expiry schedule. A saved expiry should trigger settlement. Once no expiry remains, this duty should stop querying. **Steps to reproduce** Start a claimed instance with heartbeat scheduling enabled. Leave it idle with no pending tool reviews. Observe pending-review expiry discovery in the maintenance sweep. Stacked on #17450; earlier slices are #17446 and #17443. These all reuse the delivery and transaction foundations from #15625. This PR owns pending-review expiry only. Approved-action recovery, interrupted-execution quarantine, and terminal interaction projection repair remain separate migrations. No duplicate pending-review expiry migration was found in the open PR search; credential lifetime changes in #17221 and #17274 concern the parent credential slice. ## What Changed - Replace repeated company review-deadline aggregates with an indexed single-row lookup. Add and replay a concurrent covering-index migration and generate combined snapshots throughout the stack. - Extract pending-expiry selection and settlement into its owning service. Keep provider execution and interaction publication with the gateway. - Signal deadline assignment and pending resolution inside the owning SQL transaction. Preserve outer commit/rollback and central uncertainty handling. - Register company expiry deadlines with the shared scoped coordinator. Reconstruct on startup; normal producer hints reload only their company. Preserve writes during drain and concurrent reconciliation. - Settle request expiry and invocation failure atomically. Keep interaction projection recovery in the existing separate sweep. - Remove pending-expiry scanning from the main tick. Preserve the combined manual API and startup-only behavior when scheduling is disabled. - Add a partial pending-expiry index through an idempotent concurrent migration. No queue table is added. - Document supported producers, quiet states, remaining duties, and external wake limits. ## Verification - CI run `40574661575` at `f5c935d1b89877033eb82bf890f867baeb372ef3` lost the runners for general-server shard 9/12 and E2E shard 2/8 at 20:09:03 UTC. Both job logs explicitly show a runner shutdown signal followed by cancellation; no test assertion failure was reported by those jobs. A single infrastructure retry is justified once the run completes; it has not been issued yet. Both current-head reviews are 5/5, and CI must still be green before merge. - Current master landing at `f5c935d1b89877033eb82bf890f867baeb372ef3`: 128 complete gateway expiry, startup, scoped deadline and preparation-timing cases pass with no skips. Server types and the production server build pass. Landed tracing changes from #17918 are preserved. The master migration journal prefix remains semantically identical, with only this PR's two migrations added. Fresh remote CI/review is required before merge. - Parent reconciliation at `17539e2a6cf8ac4922b3cbe06af6df361bfe5665`: 116 tests pass across the full tool gateway service, startup and scoped deadline suites against a dedicated migrated PostgreSQL database. No cases were skipped. Server typecheck passes. Fresh remote CI and review are pending. Current head: `ec28076f825e76a267f05517ff0c44a3e2283ce7`. This remains a single-duty PR stacked on #17450. - 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/ec28076f825e76a267f05517ff0c44a3e2283ce7/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 - All supported producers must share the database signal owner. Independent SQL writers and other replicas need a live notification contract; startup alone does not cover them. - A crash after authoritative expiry and before interaction projection still relies on the existing projection-repair duty. That duty, approved-action recovery, and uncertain execution recovery remain periodic in this slice. - Unknown external outcomes remain quarantined. Provider calls stay outside SQL transactions. Expiry is still checked before approving a request even if physical cleanup waits during drain. - The new index uses the existing top-level migration executor because PostgreSQL concurrent index creation cannot run in a file-wide transaction. Snapshot pruning follows the repository rule. - This establishes worker quiet and reactivation, not deployed host suspension. Existing Cloud idle-safety and external alarm coverage remain unchanged. ## 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 Latest-head CI is now green after one evidence-backed retry. Greptile is 5/5 and all review threads are resolved. --------- Co-authored-by: Paperclip --- doc/idle-tool-review-expiry.md | 155 ++++++++ ...2026-10-24-recurring-server-sleep-stack.md | 58 ++- packages/db/src/client.ts | 2 + .../0382_narrow_the_executioner.sql | 1 + .../0383_tool_review_company_expiry.sql | 1 + ...{0376_snapshot.json => 0382_snapshot.json} | 72 +++- ...{0377_snapshot.json => 0383_snapshot.json} | 94 ++++- packages/db/src/migrations/meta/_journal.json | 14 + packages/db/src/schema/tool_access.ts | 4 + .../server-startup-feedback-export.test.ts | 4 + .../__tests__/tool-gateway-service.test.ts | 352 +++++++++++++++++- server/src/index.ts | 13 +- server/src/services/company-deletion.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../src/services/issue-thread-interactions.ts | 2 + .../tool-access/activity/review-queue.ts | 5 +- server/src/services/tool-action-review.ts | 2 + server/src/services/tool-gateway.ts | 67 ++-- .../src/services/tool-review-expiry-work.ts | 25 ++ server/src/services/tool-review-expiry.ts | 65 ++++ .../tool-review-work-notifications.ts | 13 + 21 files changed, 905 insertions(+), 47 deletions(-) create mode 100644 doc/idle-tool-review-expiry.md create mode 100644 packages/db/src/migrations/0382_narrow_the_executioner.sql create mode 100644 packages/db/src/migrations/0383_tool_review_company_expiry.sql rename packages/db/src/migrations/meta/{0376_snapshot.json => 0382_snapshot.json} (99%) rename packages/db/src/migrations/meta/{0377_snapshot.json => 0383_snapshot.json} (99%) create mode 100644 server/src/services/tool-review-expiry-work.ts create mode 100644 server/src/services/tool-review-expiry.ts create mode 100644 server/src/services/tool-review-work-notifications.ts diff --git a/doc/idle-tool-review-expiry.md b/doc/idle-tool-review-expiry.md new file mode 100644 index 00000000000..d14abcbfd4a --- /dev/null +++ b/doc/idle-tool-review-expiry.md @@ -0,0 +1,155 @@ +# Pending tool-review expiry and idle behavior + +Pending tool action requests already save their review deadline in `expiresAt`. +This worker replaces the recurring pending-expiry scan with committed changes, +startup reconstruction, and that saved deadline. It does not add a queue table +or change the one-hour review policy. + +## Ownership and producer inventory + +The tool gateway owns request signing and authorization. `toolReviewExpiryService` +owns pending-expiry selection and settlement. `registerToolReviewExpiryWork` +supplies company-scoped operations to the existing shared scoped deadline worker. Startup wires +that contract without interpreting tool or provider behavior. + +The supported writers are: + +- `tool-access-policy.ts` inserts an unsigned pending request without an expiry. + It creates no expiry obligation yet. The existing signing grace remains intact. +- The normal agent gateway signing path and the board test-call path assign the + saved deadline. Each signals in its SQL transaction before that update. +- Gateway signing failures, competing-call invalidation, invalid signatures, + and legacy pending-to-approved/expired transitions signal before removing a + pending obligation. +- `commitToolActionReview` signals inside its existing human-resolution + transaction. The issue/interaction locks, human authority checks, and delivery + enqueue remain owned there. +- Administrative interaction resolution signals in the caller-owned transaction + before resolving linked requests. Review-queue invalidation signals before + invalidating unreadable or unsigned requests. +- Company deletion signals its company before removing saved review obligations. +- The expiry worker signals before its own settlement. A compare-and-set against + pending status and the observed update time protects concurrent resolution. + +Every notification requires the company ID. Normal commits reload only that +company's saved minimum expiry; due cleanup selects only that company. Startup +reconstructs all saved company deadlines once. A company with only unsigned +null-expiry requests has no deadline. + +Only the SQL mutation is wrapped. Signing, provider requests, telemetry, and +interaction publication are not moved inside an enclosing recovery transaction. +The database signal layer handles nested savepoints, outer commit/rollback, and +uncertain transaction outcomes. No recovery object is passed through callers. +Independent SQL writers and other database owners/replicas still require a live +notification contract. Startup reconstructs saved rows but is not ongoing +cross-process discovery. + +## State and recovery + +Expiry updates the authoritative request and invocation in one transaction. If +invocation settlement fails, the request remains pending and the worker retries. +After commit, the gateway reflects the result in the linked interaction. The +existing terminal projection-repair duty remains in `sweepActionReviews` and +repairs a crash or failure between authoritative settlement and projection. +That duty is a separate migration; this worker must not claim it has completed. + +The public/manual `sweepActionReviews()` still includes expiry by default. Server +startup and the recurring legacy review sweep pass `includeExpiry: false` when +expiry has its own startup pass or worker. Approved-action recovery, interrupted +execution quarantine, and terminal projection repair retain their behavior. +An unknown external outcome is never replayed by this expiry worker. + +The next deadline comes from pending rows only. Startup uses the existing +partial expiry index to reconstruct company deadlines. A company notification +uses an ordered, single-row lookup backed by the partial `(company_id, +expires_at)` index, rather than aggregating that company's pending backlog. +Null-expiry unsigned requests and retained resolved history do not keep this +worker active. The concurrent index migration uses the existing top-level +executor and is safe to apply again after interruption. + +The shared coordinator coalesces changes and preserves a write arriving during +reconciliation. Each due pass expires at most 100 rows for one company; companies +rotate between passes until the backlog drains. Failed scopes retry independently +with exponential backoff from 30 seconds to five minutes. Other companies retain +their own deadlines. New writes do not erase a failed sweep's backoff. +Once reconciliation completes with no deadline, it has no recurring SQL. Drain +retains first writes without passive database scans; resume processes overdue +reviews. Startup reconstructs persisted deadlines. Shutdown stops subscription +and timer ownership and joins active work. + +Approval still rejects an expired request synchronously before physical cleanup. +With heartbeat scheduling disabled, startup-only expiry remains as before. + +## Verification scope + +Real PostgreSQL tests record SQL to prove signing, human decisions, and due expiry +remain company-scoped. They also verify five-second and earlier replacement +deadlines, company deletion, and bounded multi-pass draining. Existing cases cover +normal gateway signing and human decisions, +configured idle and terminal history, outer commit/rollback, startup/restart, +drain/resume, concurrent producer/read, unsigned rows, multi-page expiry, and +injected invocation-write failure with atomic rollback and retry. Existing +service tests cover approval/execution, interrupted outcomes, and projection +repair through the compatible manual entry point. Startup tests cover wiring. +The external test database override requires a disposable database ending in +`_test`; the suite deletes fixtures and stops the worker before database cleanup. + +This is one worker migration, not proof that the instance can sleep. Other tool +review duties remain periodic. Existing Cloud sleep-safety gates and external +alarm coverage are unchanged. An in-process deadline cannot wake a stopped host. + +## Scoped lookup review repair (2026-10-27) + +The original scoped aggregate examined 12,004 rows to return one deadline in a +real PostgreSQL regression. The repaired lookup returns the earliest non-null +pending expiry for the requested company with `ORDER BY expires_at LIMIT 1`. +The grouped startup inventory remains unchanged. Existing company signals, +authority checks, expiry settlement, paging, backoff, and sleep gates remain +owned by their existing services. + +The additive company-expiry index is migration `0383_tool_review_company_expiry`. +It builds the index concurrently and idempotently through the existing top-level +migration executor. Initially reserving number 0402 passed per-branch checks but +caused CI to reject descendants whose migrations preceded their target branch. +A local 0380 correction fixed that ordering, but master had meanwhile added its +own 0378 and 0379 migrations. The integrated stack preserves master's history, +renumbers its unmerged migrations after 0379 in PR dependency order, and preserves +every migration's SQL bytes. Exact filename references and journal tags/indices +follow the new names. Stack journal timestamps remain ordered after master's +latest entry. Each generated snapshot includes both master and that PR's schema. + +Each PR must pass the exact target-branch migration-order gate, in addition to +the existing numbering, safety, five-snapshot retention, and schema-drift checks. +Local branch checks alone did not establish the stack's append-only ordering. +Fresh application, upgrades from the previous numbered task databases, and +replay also require verification. These unmerged stack migrations have not +changed the SQL of migrations already on master. + +Two failures shaped the repair. The first migration attempt omitted its explicit +top-level registration and PostgreSQL rejected the concurrent build inside a +transaction; adding the registration fixed application and replay. An initial +three-column index also lost to the global expiry index in the planner, which +filtered 12,001 foreign-company rows before returning a result. The final index +omits the unused request ID. With 12,000 additional pending rows in each of two +companies, including earlier foreign deadlines, all three measured reloads +examine one row. The assertion counts filtered rows as well as returned rows. +This is evidence for the recorded fixture and query plan, not a production +latency guarantee or a claim that PostgreSQL can never choose another plan. + +The complete gateway service, gateway acceptance, credential expiry, startup, +custom-image deadline, and schema snapshot suites pass: 223 tests across six +files. The run also exposed a credential test-clock overshoot; its correction +belongs to the earlier gateway PR and waits for real SQL before advancing to +an actually armed follow-up. Commit/rollback and exact expiry assertions remain. +Server typechecking passes. Combined-stack migration upgrades, full build, and +fresh remote checks remain required. Full local test:run is still unverified +because of embedded PostgreSQL startup/capacity limits; real gateway suites used +isolated, migrated test databases in the task-owned PostgreSQL instance. + + +A later review found an unbound `query` name in the SQL-capture assertion. +The due-row predicate short-circuited before that expression in the passing +fixture, and server typechecking excludes test sources. The assertion now +independently requires both the due expiry scan and the non-null deadline lookup +through each captured `read.query`. The earlier passing run did not detect this +name error; acceptance must include the service suite with both assertions. 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 009401a1d93..5bdacbcd130 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -18,7 +18,7 @@ Do not mark an item done from a timer removal or a green check alone. | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | -| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged after CI and passing reviews; gateway credential expiry in progress; tool reviews pending | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry implemented on slice 04; other review duties pending | | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | @@ -61,3 +61,59 @@ connector transport architecture, probes, telemetry/exporter qualification, and end-to-end provider suspension are outside this requested recurring-server stack; none are claimed solved by these PRs. Required active-operation lease renewals and request timeouts remain with their owners. + +## Next owner audit: tool reviews + +Inspected after gateway credential slice `75332b9641`. The existing +`sweepActionReviews` combines four reasons that need separate migration slices: + +1. Pending review expiry: `status = pending` and saved `expiresAt`. +2. Approved action recovery: approved rows that still owe execution/settlement. +3. Interrupted execution: executing rows with the existing ten-minute stale + threshold; unknown provider outcomes must remain quarantined, not replayed. +4. Interaction projection repair: terminal requests whose interaction result + does not reflect the authoritative request/invocation state. + +The next branch is `codex/sleep-04-tool-review-expiry`. Implementation now separates pending expiry with commit-aware signals and its +existing deadline. Preserve the other three duties while migrating the first, then +retire them individually. Keep the public/manual recovery entry point compatible +or update all callers and tests explicitly; tests currently call the combined +method to cause expiry. + +Producer locations found by the audit: + +- `tool-access-policy.ts`: creates pending requests with null signature/expiry. +- `tool-gateway.ts`: signs pending requests and assigns their one-hour expiry; + handles competing-call invalidation, approval/execution transitions, provider + outcomes, and sweep settlement. Audit all writes, not just initial insertion. +- `tool-action-review.ts`: owns a transaction for human resolution. +- `tool-access/activity/review-queue.ts`: invalidates unsigned/unreadable requests. +- `issue-thread-interactions.ts`: resolves linked pending/approved requests when + their governing card is resolved administratively. + +The unsigned-row signing grace is distinct from the saved review expiry. Do not +invent a new expiry policy while removing the scan. A pending row can be observed +between insert and signing; expiry workers must not cancel an in-flight producer. +Keep provider execution outside database transactions. Signals must be registered +before writes in existing transaction ownership, without wrapping provider calls. + +Latest evidence: #17443 has all checks passed and both review threads resolved. +#17446 at `d41590ac0e` has all checks passed, Greptile 5/5, and no review threads. +#17450 at `c5c53fc43f` fixes the deadline query cost with bounded indexed lookups +and an idempotent concurrent session index. Both findings are resolved; fresh +Greptile is 5/5 and CI is running. Local migration/replay, 47 expiry/startup tests, +workspace typecheck/build passed. A synthetic 40,000-row PostgreSQL plan reads +one row from each expiry index. Recheck external state before readiness claims. + +Slice 04 now has 11 real-PostgreSQL worker lifecycle tests plus 59 existing +service tests and 39 startup tests passing (109 total). The initial fixture +incorrectly treated an argument-summary object as a string. After correction, +the failure test exposed its invalid expectation of no idle hold during an +unresolved write; it now checks rollback while retaining the hold. A combined +run also exposed fixture cleanup ordering: this repository uses ordered hooks, +so the worker must stop before the outer hook deletes rows. The corrected full +selection passes. The current implementation adds a partial pending-expiry +index, applies/replays its migration, and preserves approval-time rejection and +separate approved/executing/projection recovery. Workspace typecheck/build and all 92 existing gateway acceptance tests also +passed. The first acceptance command matched no files; the corrected +`tool-gateway.test.ts` selection passed. Fresh PR CI/review remains required. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 77fb0c62fa1..df20aa973ab 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -502,6 +502,8 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0345_luxuriant_carmella_unuscione.sql", "0369_late_logan.sql", "0381_mean_scream.sql", + "0382_narrow_the_executioner.sql", + "0383_tool_review_company_expiry.sql", ]); async function applyPendingMigrationsManually( diff --git a/packages/db/src/migrations/0382_narrow_the_executioner.sql b/packages/db/src/migrations/0382_narrow_the_executioner.sql new file mode 100644 index 00000000000..af2e60c65f7 --- /dev/null +++ b/packages/db/src/migrations/0382_narrow_the_executioner.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_action_requests_pending_expiry_idx" ON "tool_action_requests" USING btree ("expires_at","id") WHERE "tool_action_requests"."status" = 'pending'; diff --git a/packages/db/src/migrations/0383_tool_review_company_expiry.sql b/packages/db/src/migrations/0383_tool_review_company_expiry.sql new file mode 100644 index 00000000000..75a364ecdf1 --- /dev/null +++ b/packages/db/src/migrations/0383_tool_review_company_expiry.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_action_requests_company_pending_expiry_idx" ON "tool_action_requests" USING btree ("company_id","expires_at") WHERE "tool_action_requests"."status" = 'pending'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0376_snapshot.json b/packages/db/src/migrations/meta/0382_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0376_snapshot.json rename to packages/db/src/migrations/meta/0382_snapshot.json index 1015edaf69f..6c27dd9aa22 100644 --- a/packages/db/src/migrations/meta/0376_snapshot.json +++ b/packages/db/src/migrations/meta/0382_snapshot.json @@ -1,6 +1,6 @@ { - "id": "7832df38-fa1b-4687-bd06-eccbc29bd28c", - "prevId": "3e17203b-ec95-41e9-8363-685982450199", + "id": "ecec138e-5312-4ad6-b051-bd4ee9ed4e34", + "prevId": "2e93deb7-f24c-42d1-8c8e-30b21cff90bc", "version": "7", "dialect": "postgresql", "tables": { @@ -47782,6 +47782,18 @@ "primaryKey": false, "notNull": false }, + "git_authorization_grant_id": { + "name": "git_authorization_grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "git_authorized_by_user_id": { + "name": "git_authorized_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "pause_reason": { "name": "pause_reason", "type": "text", @@ -47920,6 +47932,19 @@ ], "onDelete": "no action", "onUpdate": "no action" + }, + "projects_git_authorization_grant_id_connection_grants_id_fk": { + "name": "projects_git_authorization_grant_id_connection_grants_id_fk", + "tableFrom": "projects", + "tableTo": "connection_grants", + "columnsFrom": [ + "git_authorization_grant_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" } }, "compositePrimaryKeys": {}, @@ -52787,6 +52812,28 @@ "method": "btree", "with": {} }, + "tool_action_requests_pending_expiry_idx": { + "name": "tool_action_requests_pending_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -55197,6 +55244,21 @@ "method": "btree", "with": {} }, + "tool_gateway_sessions_expiry_idx": { + "name": "tool_gateway_sessions_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_gateway_sessions_run_idx": { "name": "tool_gateway_sessions_run_idx", "columns": [ @@ -55942,6 +56004,12 @@ "notNull": true, "default": "''" }, + "responsible_user_id": { + "name": "responsible_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "owner_note": { "name": "owner_note", "type": "text", diff --git a/packages/db/src/migrations/meta/0377_snapshot.json b/packages/db/src/migrations/meta/0383_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0377_snapshot.json rename to packages/db/src/migrations/meta/0383_snapshot.json index 6e89edc3a9b..3336a6bb20e 100644 --- a/packages/db/src/migrations/meta/0377_snapshot.json +++ b/packages/db/src/migrations/meta/0383_snapshot.json @@ -1,6 +1,6 @@ { - "id": "a0fcd9b8-f125-4985-8795-f567466b9aef", - "prevId": "7832df38-fa1b-4687-bd06-eccbc29bd28c", + "id": "410457fc-266b-4bd0-8bc0-2b14b0634d07", + "prevId": "ecec138e-5312-4ad6-b051-bd4ee9ed4e34", "version": "7", "dialect": "postgresql", "tables": { @@ -47782,6 +47782,18 @@ "primaryKey": false, "notNull": false }, + "git_authorization_grant_id": { + "name": "git_authorization_grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "git_authorized_by_user_id": { + "name": "git_authorized_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "pause_reason": { "name": "pause_reason", "type": "text", @@ -47920,6 +47932,19 @@ ], "onDelete": "no action", "onUpdate": "no action" + }, + "projects_git_authorization_grant_id_connection_grants_id_fk": { + "name": "projects_git_authorization_grant_id_connection_grants_id_fk", + "tableFrom": "projects", + "tableTo": "connection_grants", + "columnsFrom": [ + "git_authorization_grant_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" } }, "compositePrimaryKeys": {}, @@ -52787,6 +52812,50 @@ "method": "btree", "with": {} }, + "tool_action_requests_pending_expiry_idx": { + "name": "tool_action_requests_pending_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_company_pending_expiry_idx": { + "name": "tool_action_requests_company_pending_expiry_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -55197,6 +55266,21 @@ "method": "btree", "with": {} }, + "tool_gateway_sessions_expiry_idx": { + "name": "tool_gateway_sessions_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_gateway_sessions_run_idx": { "name": "tool_gateway_sessions_run_idx", "columns": [ @@ -55942,6 +56026,12 @@ "notNull": true, "default": "''" }, + "responsible_user_id": { + "name": "responsible_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "owner_note": { "name": "owner_note", "type": "text", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 176a963f7a4..02023247485 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2661,6 +2661,20 @@ "when": 1793060791725, "tag": "0381_mean_scream", "breakpoints": true + }, + { + "idx": 382, + "version": "7", + "when": 1793060791726, + "tag": "0382_narrow_the_executioner", + "breakpoints": true + }, + { + "idx": 383, + "version": "7", + "when": 1793060791727, + "tag": "0383_tool_review_company_expiry", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/tool_access.ts b/packages/db/src/schema/tool_access.ts index 825eb91589f..e970c445b6c 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -832,6 +832,10 @@ export const toolActionRequests = pgTable( }, (table) => [ index("tool_action_requests_company_status_idx").on(table.companyId, table.status), + index("tool_action_requests_pending_expiry_idx").on(table.expiresAt, table.id) + .where(sql`${table.status} = 'pending'`), + index("tool_action_requests_company_pending_expiry_idx").on(table.companyId, table.expiresAt) + .where(sql`${table.status} = 'pending'`), index("tool_action_requests_invocation_idx").on(table.invocationId), index("tool_action_requests_issue_idx").on(table.companyId, table.issueId), ], diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 9788b8599ff..2ede1c6c85f 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -70,6 +70,10 @@ const { }, toolGateway: { sweepActionReviews: vi.fn(async () => ({ scanned: 0 })), + sweepExpiredActionReviews: vi.fn(async () => ({ scanned: 0 })), + nextActionReviewExpiryAt: vi.fn(async () => null), + actionReviewExpiryDeadlines: vi.fn(async () => []), + sweepExpiredActionReviewsForCompany: vi.fn(async () => ({ scanned: 0 })), cleanupExpiredSessions: vi.fn(async () => ({ deletedCount: 0 })), nextCredentialExpiryAt: vi.fn(async () => null), }, diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index cc804845ff6..3e9829670c6 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -1,11 +1,12 @@ +import { deleteCompany } from "../services/company-deletion.js"; import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { and, eq } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { activityLog, agentWakeupRequests, @@ -54,7 +55,13 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerToolReviewExpiryWork } from "../services/tool-review-expiry-work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_REVIEW_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; const testToolActionSigningSecret = "test-tool-action-signing-secret"; type ToolGatewayServiceOptions = NonNullable[1]>; @@ -203,13 +210,22 @@ function runSessionInput(company: IdRef, agent: IdRef, run: IdRef): Parameters { let db!: ReturnType; let tempDb: Awaited> | null = null; + let stopExpiryWorker: (() => Promise) | undefined; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-"); - db = createDb(tempDb.connectionString); + if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-"); + db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { + maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0, + }); + await clearFixtures(); }, 20_000); afterEach(async () => { + await stopExpiryWorker?.(); stopExpiryWorker = undefined; + await clearFixtures(); + }); + + async function clearFixtures() { vi.unstubAllEnvs(); await db.delete(activityLog); await db.delete(agentWakeupRequests); @@ -232,12 +248,336 @@ describeEmbeddedPostgres("tool gateway service", () => { await db.delete(agents); await db.delete(companyMemberships); await db.delete(companies); - }); + } afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); + describe("pending review expiry worker", () => { + let coordinator: ReturnType | undefined; + let gateway: ReturnType; + let fixture: Awaited>; + let draining: boolean; + const errors: unknown[] = []; + + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + fixture = await createRunFixture(db); + await db.insert(toolPolicies).values(reviewNotePolicyRow(fixture.company)); + gateway = createTestToolGatewayService(db); + draining = false; errors.length = 0; + stopExpiryWorker = stopWorker; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + async function stopWorker() { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.execute(sql`drop trigger if exists sleep_review_fail on tool_invocations`); + await db.execute(sql`drop function if exists sleep_review_fail()`); + expect(errors).toEqual([]); + } + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerToolReviewExpiryWork(coordinator, gateway, () => !draining).ready; + } + async function settle(ms = 60_000) { + await vi.advanceTimersByTimeAsync(ms); + if (!draining) await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 }); + } + async function quiet() { + await settle(); expect(coordinator!.nextWakeAt()).toBeNull(); + const selects = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute"); + await settle(60 * 60_000); + expect(selects).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); + selects.mockRestore(); executes.mockRestore(); + } + async function request(service = gateway, source = db, context = fixture) { + const before = new Set((await source.select({ id: toolActionRequests.id }).from(toolActionRequests) + .where(eq(toolActionRequests.companyId, context.company.id))).map(row => row.id)); + const session = await service.createSession(runSessionInput(context.company, context.agent, context.run)); + const noteId = randomUUID(); + await expect(service.executeTool({ sessionToken: session.token, tool: "mcp-remote-fixture:update_note", + parameters: { noteId, body: "Awaiting a human" } })).rejects.toMatchObject({ reasonCode: "approval_required" }); + const rows = (await source.select().from(toolActionRequests).where(eq(toolActionRequests.companyId, context.company.id))) + .filter(row => !before.has(row.id)); + expect(rows).toHaveLength(1); + return rows[0]; + } + async function expectExpired(id: string, invocationId: string) { + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, id)))[0].status).toBe("expired"); + expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, invocationId)))[0]) + .toMatchObject({ status: "failed", errorCode: "action_expired" }); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); + } + + async function requestWithExpiry(expiresAt: Date, context = fixture) { + return db.transaction(async tx => { + const source = tx as unknown as typeof db; + const pending = await request(createTestToolGatewayService(source), source, context); + const [saved] = await tx.update(toolActionRequests).set({ expiresAt }) + .where(eq(toolActionRequests.id, pending.id)).returning(); + return saved; + }); + } + + it("keeps signing and human resolution reads scoped to their company", async () => { + const other = await createRunFixture(db); + await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company)); + const foreign = await request(gateway, db, other); + await start(); await settle(1); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query, params) => { + if (/^select/i.test(query) && query.includes('"tool_action_requests"') && + (query.includes('min(') || query.includes('"expires_at" <=') || query.includes('"expires_at" is not null'))) reads.push({ query, params: [...params] }); + }; + try { + for (let index = 0; index < 3; index++) { + const pending = await request(); await settle(1000); + await commitToolActionReview(db, { companyId: fixture.company.id, actionRequestId: pending.id, + decision: "rejected", actor: { userId: "reviewer" } }); + await settle(1000); + } + } finally { db.$client.options.debug = debug; } + expect(reads.length).toBeGreaterThan(0); + for (const read of reads) { + expect(read.query).toContain('"tool_action_requests"."company_id" ='); + expect(read.params).toContain(fixture.company.id); + expect(read.params).not.toContain(other.company.id); + } + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, foreign.id)))[0].status).toBe("pending"); + expect(coordinator!.nextWakeAt()).toBe(foreign.expiresAt!.getTime()); + }); + + it("reads one indexed deadline from a large company backlog on repeated reloads", async () => { + const pending = await request(); + const foreignContext = await createRunFixture(db); + await db.insert(toolPolicies).values(reviewNotePolicyRow(foreignContext.company)); + const foreign = await requestWithExpiry(new Date(Date.now() + 1000), foreignContext); + await db.execute(sql`insert into tool_action_requests + (company_id, invocation_id, canonical_arguments_hash, canonical_arguments_summary, expires_at) + select ${fixture.company.id}, ${pending.invocationId}, 'backlog', '{}'::jsonb, + ${pending.expiresAt!.toISOString()}::timestamptz + n * interval '1 second' + from generate_series(1, 12000) n`); + await db.execute(sql`insert into tool_action_requests + (company_id, invocation_id, canonical_arguments_hash, canonical_arguments_summary, expires_at) + select ${foreignContext.company.id}, ${foreign.invocationId}, 'foreign backlog', '{}'::jsonb, + ${foreign.expiresAt!.toISOString()}::timestamptz - n * interval '1 second' + from generate_series(1, 12000) n`); + await db.insert(toolActionRequests).values([ + { ...pending, id: randomUUID(), interactionId: null, expiresAt: null }, + { ...pending, id: randomUUID(), interactionId: null, status: "rejected", expiresAt: new Date(0) }, + ]); + await db.execute(sql`analyze tool_action_requests`); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query, params) => { + if (/^select/i.test(query) && query.includes('"tool_action_requests"')) reads.push({ query, params: [...params] }); + }; + try { + for (let i = 0; i < 3; i++) { + expect(await gateway.actionReviewExpiryDeadlines(fixture.company.id)).toEqual([ + { companyId: fixture.company.id, dueAt: pending.expiresAt!.getTime() }, + ]); + } + } finally { db.$client.options.debug = debug; } + expect(reads).toHaveLength(3); + type Plan = { "Node Type": string; "Actual Rows": number; "Actual Loops": number; + "Rows Removed by Filter"?: number; "Rows Removed by Index Recheck"?: number; Plans?: Plan[] }; + const scannedRows = (plan: Plan): number => + (/Scan/.test(plan["Node Type"]) ? (plan["Actual Rows"] + (plan["Rows Removed by Filter"] ?? 0) + + (plan["Rows Removed by Index Recheck"] ?? 0)) * plan["Actual Loops"] : 0) + + (plan.Plans ?? []).reduce((count, child) => count + scannedRows(child), 0); + for (const read of reads) { + const [result] = await db.$client.unsafe( + `explain (analyze, buffers, format json) ${read.query}`, + read.params as Parameters[1], + ); + const plan = result["QUERY PLAN"][0].Plan as Plan; + expect(scannedRows(plan), JSON.stringify(plan)).toBeLessThanOrEqual(1); + } + expect(await gateway.actionReviewExpiryDeadlines(randomUUID())).toEqual([]); + expect(await gateway.actionReviewExpiryDeadlines()).toEqual(expect.arrayContaining([ + { companyId: fixture.company.id, dueAt: pending.expiresAt!.getTime() }, + { companyId: foreignContext.company.id, dueAt: foreign.expiresAt!.getTime() - 12_000_000 }, + ])); + }); + + it("preserves a five-second deadline and an earlier committed replacement", async () => { + await start(); await quiet(); + const later = await requestWithExpiry(new Date(Date.now() + 120_000)); await settle(1); + expect(coordinator!.nextWakeAt()).toBe(later.expiresAt!.getTime()); + const earlier = await requestWithExpiry(new Date(Date.now() + 5000)); await settle(1); + expect(coordinator!.nextWakeAt()).toBe(earlier.expiresAt!.getTime()); + await settle(6000); await expectExpired(earlier.id, earlier.invocationId); + expect(coordinator!.nextWakeAt()).toBe(later.expiresAt!.getTime()); + await settle(120_000); await expectExpired(later.id, later.invocationId); await quiet(); + }); + + it("limits due expiry reads to the company whose deadline arrived", async () => { + const other = await createRunFixture(db); + await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company)); + const foreign = await request(gateway, db, other); + const due = await requestWithExpiry(new Date(Date.now() + 5000)); + await start(); await settle(1); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query, params) => { + if (/^select/i.test(query) && query.includes('"tool_action_requests"') && + (query.includes('min(') || query.includes('"expires_at" <=') || query.includes('"expires_at" is not null'))) reads.push({ query, params: [...params] }); + }; + try { await settle(6000); } finally { db.$client.options.debug = debug; } + expect(reads.some(read => read.query.includes('"expires_at" <='))).toBe(true); + expect(reads.some(read => read.query.includes('"expires_at" is not null'))).toBe(true); + for (const read of reads) { + expect(read.query).toContain('"tool_action_requests"."company_id" ='); + expect(read.params).toContain(fixture.company.id); + expect(read.params).not.toContain(other.company.id); + } + await expectExpired(due.id, due.invocationId); + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, foreign.id)))[0].status).toBe("pending"); + expect(coordinator!.nextWakeAt()).toBe(foreign.expiresAt!.getTime()); + }); + + it("removes cached review deadlines after company deletion commits", async () => { + await request(); await start(); await settle(1); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.run.id)); + await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" }) + .where(eq(agents.id, fixture.agent.id)); + expect(await deleteCompany(db, fixture.company.id)).toMatchObject({ id: fixture.company.id }); + await quiet(); + expect(await gateway.actionReviewExpiryDeadlines()).toEqual([]); + }); + + it("is quiet when configured idle, wakes on signing, expires the review and card, then stays quiet with history", async () => { + await start(); await quiet(); + const pending = await request(); + await settle(); expect(coordinator!.nextWakeAt()).toBe(pending.expiresAt!.getTime()); + const reads = vi.spyOn(db, "select"); + await settle(5 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await settle(60 * 60_000); + await expectExpired(pending.id, pending.invocationId); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!)))[0].status).toBe("expired"); + await quiet(); + }); + + it.each(["approved", "rejected"] as const)("disarms when a human resolves the last review as %s", async decision => { + await start(); const pending = await request(); await settle(); + await commitToolActionReview(db, { companyId: fixture.company.id, actionRequestId: pending.id, + decision, actor: { userId: "reviewer" } }); + await quiet(); + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe(decision); + }); + + it.each([false, true])("observes outer transaction settlement (rollback %s)", async rollback => { + await start(); await quiet(); + let release!: () => void, staged!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { staged = resolve; }); + const aborted = new Error("rollback fixture"); + const writing = db.transaction(async tx => { + const scoped = tx as unknown as typeof db; + await request(createTestToolGatewayService(scoped), scoped); + staged(); await gate; + if (rollback) throw aborted; + }).catch(error => { if (error !== aborted) throw error; }); + await Promise.race([ready, writing]); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(await db.select().from(toolActionRequests)).toEqual([]); + } finally { release(); } + await writing; await settle(); + if (rollback) await quiet(); + else { + expect(coordinator!.nextWakeAt()).toBe(await gateway.nextActionReviewExpiryAt()); + await settle(60 * 60_000); await quiet(); + } + }); + + it("reconstructs saved expiry on restart", async () => { + const pending = await request(); + await start(); await settle(); + expect(coordinator!.nextWakeAt()).toBe(pending.expiresAt!.getTime()); + await coordinator!.stop(); coordinator = undefined; + await settle(60 * 60_000); + await start(); await settle(); + await expectExpired(pending.id, pending.invocationId); await quiet(); + }); + + it("retains the first write through drain and enforces approval expiry before cleanup resumes", async () => { + await start(); await quiet(); draining = true; + const pending = await request(); await settle(); + const reads = vi.spyOn(db, "select"); + await settle(2 * 60 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + await expect(commitToolActionReview(db, { companyId: fixture.company.id, actionRequestId: pending.id, + decision: "approved", actor: { userId: "reviewer" } })).rejects.toThrow("This review has expired"); + draining = false; await settle(); + await expectExpired(pending.id, pending.invocationId); await quiet(); + }); + + it("preserves a signing write arriving during an empty deadline read", async () => { + let release!: () => void, reading!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { reading = resolve; }); + const actual = gateway.actionReviewExpiryDeadlines.bind(gateway); + vi.spyOn(gateway, "actionReviewExpiryDeadlines").mockImplementationOnce(async () => { + const deadline = await actual(); reading(); await gate; return deadline; + }); + const starting = start(); await ready; + let pending: Awaited>; + try { pending = await request(); } finally { release(); } + await starting; await settle(); + expect(coordinator!.nextWakeAt()).toBe(pending!.expiresAt!.getTime()); + }); + + it("leaves unsigned requests without a saved expiry alone", async () => { + const pending = await request(); + await db.update(toolActionRequests).set({ expiresAt: null, signedArguments: null }) + .where(eq(toolActionRequests.id, pending.id)); + await start(); await quiet(); + expect((await db.select().from(toolActionRequests))[0].status).toBe("pending"); + }); + + it("expires all overdue pages without replaying approved or interrupted actions", async () => { + const pending = await request(); + const past = new Date(Date.now() - 1000); + await db.update(toolActionRequests).set({ expiresAt: past }).where(eq(toolActionRequests.id, pending.id)); + await db.insert(toolActionRequests).values(Array.from({ length: 101 }, () => ({ + ...pending, id: randomUUID(), interactionId: null, expiresAt: past, + }))); + const passes = vi.spyOn(gateway, "sweepExpiredActionReviewsForCompany"); + await start(); await quiet(); + const results = await Promise.all(passes.mock.results.map(result => result.value)); + expect(results.length).toBeGreaterThan(1); + expect(results.every(result => result.scanned <= 100)).toBe(true); + const rows = await db.select().from(toolActionRequests); + expect(rows).toHaveLength(102); + expect(rows.every(row => row.status === "expired")).toBe(true); + }); + + it("rolls expiry back when invocation settlement fails, then retries the complete transition", async () => { + const pending = await request(); await start(); await settle(); + await db.execute(sql`create function sleep_review_fail() returns trigger language plpgsql as $$ + begin raise exception 'expiry settlement fixture failure'; end $$`); + await db.execute(sql`create trigger sleep_review_fail before update on tool_invocations + for each row when (new.error_code = 'action_expired') execute function sleep_review_fail()`); + await vi.advanceTimersByTimeAsync(60 * 60_000); + await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("pending"); + expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0].status).toBe("awaiting_approval"); + await db.execute(sql`drop trigger sleep_review_fail on tool_invocations`); + expect(errors.every(error => error instanceof AggregateError && error.errors.every( + failure => (failure as { cause?: Error }).cause?.message === "expiry settlement fixture failure"))).toBe(true); + errors.length = 0; + await settle(); await expectExpired(pending.id, pending.invocationId); await quiet(); + }); + }); + it("gates write tools with an action request and executes only stored reviewed arguments once", async () => { const { company, agent, run } = await createRunFixture(db); await db.insert(toolPolicies).values(reviewNotePolicyRow(company)); diff --git a/server/src/index.ts b/server/src/index.ts index 4116e46bd93..ea24010565f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerToolReviewExpiryWork } from "./services/tool-review-expiry-work.js"; import { registerGatewayCredentialExpiryWork } from "./services/gateway-credential-expiry-work.js"; import { registerSecretProposalExpiryWork } from "./services/secret-proposal-expiry-work.js"; import { registerCustomImageCleanupWork } from "./services/custom-image-cleanup-work.js"; @@ -1464,7 +1465,15 @@ async function startServerWithDatabaseTeardown( })); }; - await timeStartupRecovery("tool_reviews", () => app.locals.toolGateway.sweepActionReviews()).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed")); + await timeStartupRecovery("tool_reviews", async () => { + if (config.heartbeatSchedulerEnabled) { + await registerToolReviewExpiryWork(deliveryWork, app.locals.toolGateway, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + } else { + await app.locals.toolGateway.sweepExpiredActionReviews(); + } + return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false }); + }).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed")); await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled ? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready @@ -1894,7 +1903,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "periodic tool connection health sweep failed"); })); - trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); + trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false }).catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 7e4a41a95d7..ab549ecbfb8 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { agentObservabilityService } from "./agent-observability.js"; import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server"; @@ -62,6 +63,7 @@ export async function deleteCompany(db: Db, id: string) { if (!existing) return null; await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id); + await notifyToolReviewChange(tx, id); await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id); // Finance can reference costs and both can reference runs. Incidents // reference policies and approvals; delete these dependents first. diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 191b8aef36b..210c830f91a 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", + toolReviewExpiry: "tool-review-expiry", gatewayCredentialExpiry: "gateway-credential-expiry", secretProposalExpiry: "secret-proposal-expiry", customImageExpiry: "custom-image-expiry", diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 74e8d8216db..48ecf1d9dbe 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,3 +1,4 @@ +import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { normalizeEscapedLineBreaks } from "@paperclipai/shared/validators/text"; import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { historicalQuestionCondition } from "./issue-question-context.js"; @@ -1083,6 +1084,7 @@ async function resolveLinkedToolActionRequests( }, ) { if (interaction.kind !== "request_confirmation") return; + await notifyToolReviewChange(db, interaction.companyId); await db .update(toolActionRequests) .set({ diff --git a/server/src/services/tool-access/activity/review-queue.ts b/server/src/services/tool-access/activity/review-queue.ts index 2eb5d1e4c9e..37df9bb6c94 100644 --- a/server/src/services/tool-access/activity/review-queue.ts +++ b/server/src/services/tool-access/activity/review-queue.ts @@ -1,3 +1,4 @@ +import { writeToolReviewChange } from "../../tool-review-work-notifications.js"; import { and, desc, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { toolApplications, toolActionRequests, toolCatalogEntries, toolConnections, toolInvocations } from "@paperclipai/db"; @@ -86,7 +87,7 @@ listActionRequests: async ( if (!readable) invalidRequestIds.push(request.id); } if (invalidRequestIds.length > 0) { - await db + await writeToolReviewChange(db, companyId, tx => tx .update(toolActionRequests) .set({ status: "cancelled", @@ -99,7 +100,7 @@ listActionRequests: async ( eq(toolActionRequests.status, "pending"), inArray(toolActionRequests.id, invalidRequestIds), ), - ); + )); } const hiddenIds = new Set([ ...invalidRequestIds, diff --git a/server/src/services/tool-action-review.ts b/server/src/services/tool-action-review.ts index a1735487f0b..664b4cdc194 100644 --- a/server/src/services/tool-action-review.ts +++ b/server/src/services/tool-action-review.ts @@ -1,3 +1,4 @@ +import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { notifyDeliveryWork, DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { and, eq } from "drizzle-orm"; import { @@ -140,6 +141,7 @@ export async function commitToolActionReview( throw conflict("This review has already been resolved"); } const now = new Date(); + await notifyToolReviewChange(tx, current.companyId); const [updated] = await tx .update(toolActionRequests) .set({ diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index c75413e8529..2ff322ad308 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,3 +1,5 @@ +import { toolReviewExpiryService } from "./tool-review-expiry.js"; +import { writeToolReviewChange } from "./tool-review-work-notifications.js"; import { restConnectorFor } from "./connectors/rest-registry.js"; import type { runPromptInjectionProvider } from "./prompt-injection-provider.js"; import { promptInjectionPreventionService } from "./prompt-injection-prevention.js"; @@ -2569,7 +2571,7 @@ export function createToolGatewayService( signingSecret: options.toolActionSigningSecret, }); } catch (error) { - await db + await writeToolReviewChange(db, actionRequest.companyId, tx => tx .update(toolActionRequests) .set({ status: "cancelled", @@ -2581,7 +2583,7 @@ export function createToolGatewayService( eq(toolActionRequests.id, actionRequest.id), eq(toolActionRequests.status, "pending"), ), - ); + )); if (error instanceof ToolActionSigningSecretMissingError) { await db .update(toolInvocations) @@ -2728,7 +2730,7 @@ export function createToolGatewayService( // an expired row. When the guard matches no row, the row is already resolved. // Do not emit an approval card for a dead row; throw a retriable conflict so // the retry finds the live request or creates a fresh, signable one. - const signedRows = await db + const signedRows = await writeToolReviewChange(db, actionRequest.companyId, tx => tx .update(toolActionRequests) .set({ interactionId: interaction.id, @@ -2746,7 +2748,7 @@ export function createToolGatewayService( eq(toolActionRequests.status, "pending"), ), ) - .returning({ id: toolActionRequests.id }); + .returning({ id: toolActionRequests.id })); if (signedRows.length === 0) { await db .update(toolInvocations) @@ -8667,7 +8669,7 @@ export function createToolGatewayService( pendingRequest.expiresAt.getTime() <= Date.now(); if (pendingUnsigned || pendingExpired) { const now = new Date(); - await db + await writeToolReviewChange(db, pendingRequest.companyId, tx => tx .update(toolActionRequests) .set({ status: "expired", resolvedAt: now, updatedAt: now }) .where( @@ -8675,7 +8677,7 @@ export function createToolGatewayService( eq(toolActionRequests.id, match.actionRequest.id), eq(toolActionRequests.status, "pending"), ), - ); + )); await db .update(toolInvocations) .set({ @@ -8878,6 +8880,11 @@ export function createToolGatewayService( }; } + const reviewExpiry = toolReviewExpiryService(db, async expired => { + void emitConnectionInvoked(db, expired.invocationId); + await reflectToolActionInteractionLifecycle(expired); + }); + return { async recordRuntimeMcpDeliveryDiagnostic(input: { companyId: string; @@ -9730,6 +9737,7 @@ export function createToolGatewayService( }, ); } + const pendingActionRequest = recorded.actionRequest; const canonicalArguments = canonicalToolArguments(requestedParameters); const canonicalArgumentsHash = toolRequestIdentityHash(requestedParameters, options.toolActionSigningSecret); const approvalSnapshot = await connectedRemoteApprovalSnapshot( @@ -9750,7 +9758,7 @@ export function createToolGatewayService( signingSecret: options.toolActionSigningSecret, }); } catch (error) { - await db + await writeToolReviewChange(db, pendingActionRequest.companyId, tx => tx .update(toolActionRequests) .set({ status: "cancelled", @@ -9759,10 +9767,10 @@ export function createToolGatewayService( }) .where( and( - eq(toolActionRequests.id, recorded.actionRequest.id), + eq(toolActionRequests.id, pendingActionRequest.id), eq(toolActionRequests.status, "pending"), ), - ); + )); if (error instanceof ToolActionSigningSecretMissingError) { await db .update(toolInvocations) @@ -9791,7 +9799,7 @@ export function createToolGatewayService( tool, argumentsSummary: argumentValidation.summary, }); - await db + await writeToolReviewChange(db, pendingActionRequest.companyId, tx => tx .update(toolActionRequests) .set({ canonicalArgumentsHash, @@ -9801,7 +9809,7 @@ export function createToolGatewayService( expiresAt: new Date(Date.now() + 60 * 60 * 1000), updatedAt: new Date(), }) - .where(eq(toolActionRequests.id, recorded.actionRequest.id)); + .where(eq(toolActionRequests.id, pendingActionRequest.id))); await writeToolCallEvent({ invocationId, actionRequestId: recorded.actionRequest.id, @@ -9964,7 +9972,12 @@ export function createToolGatewayService( return buildTestCallStatus(actionRequest, invocation); }, - async sweepActionReviews() { + sweepExpiredActionReviews: reviewExpiry.sweep, + nextActionReviewExpiryAt: reviewExpiry.next, + actionReviewExpiryDeadlines: reviewExpiry.deadlines, + sweepExpiredActionReviewsForCompany: reviewExpiry.sweepForCompany, + + async sweepActionReviews(input: { includeExpiry?: boolean } = {}) { // A process may stop between persisting a provider outcome and updating the // feed projection. Reconcile from authoritative rows before delivering it. const unreflected = await db @@ -10002,7 +10015,7 @@ export function createToolGatewayService( const now = new Date(); const staleAt = new Date(now.getTime() - 10 * 60_000); let cursor: string | undefined; - let scanned = 0; + let scanned = input.includeExpiry === false ? 0 : (await reviewExpiry.sweep(now)).scanned; for (;;) { const rows = await db .select() @@ -10010,10 +10023,6 @@ export function createToolGatewayService( .where( and( or( - and( - eq(toolActionRequests.status, "pending"), - lte(toolActionRequests.expiresAt, now), - ), eq(toolActionRequests.status, "approved"), and( eq(toolActionRequests.status, "executing"), @@ -10039,15 +10048,9 @@ export function createToolGatewayService( ); continue; } - const status = row.status === "pending" ? "expired" : "failed"; - const errorCode = - status === "failed" - ? "tool_execution_outcome_unknown" - : "action_expired"; - const errorMessage = - status === "failed" - ? "Execution was interrupted; the external outcome is unknown. Inspect the provider before retrying." - : "The approval request expired before a decision."; + const status = "failed"; + const errorCode = "tool_execution_outcome_unknown"; + const errorMessage = "Execution was interrupted; the external outcome is unknown. Inspect the provider before retrying."; const [changed] = await db .update(toolActionRequests) .set({ status, resolvedAt: now, updatedAt: now }) @@ -10187,7 +10190,7 @@ export function createToolGatewayService( } if (!signedPayload) { if (actionRequest.status === "pending") { - await db + await writeToolReviewChange(db, actionRequest.companyId, tx => tx .update(toolActionRequests) .set({ status: "cancelled", @@ -10199,7 +10202,7 @@ export function createToolGatewayService( eq(toolActionRequests.id, actionRequest.id), eq(toolActionRequests.status, "pending"), ), - ); + )); } await reflectToolActionInteractionLifecycle({ actionRequestId: actionRequest.id, @@ -10764,7 +10767,7 @@ export function createToolGatewayService( actionRequest.expiresAt.getTime() <= Date.now() ) { const expiredAt = new Date(); - const [expired] = await db + const [expired] = await writeToolReviewChange(db, actionRequest.companyId, tx => tx .update(toolActionRequests) .set({ status: "expired", @@ -10777,7 +10780,7 @@ export function createToolGatewayService( inArray(toolActionRequests.status, ["pending", "approved"]), ), ) - .returning({ id: toolActionRequests.id }); + .returning({ id: toolActionRequests.id })); if (expired) { await reflectToolActionInteractionLifecycle({ actionRequestId: expired.id, @@ -10812,7 +10815,7 @@ export function createToolGatewayService( interaction?.kind === "request_confirmation" && interaction.status === "accepted" ) { - const [approved] = await db + const [approved] = await writeToolReviewChange(db, actionRequest.companyId, tx => tx .update(toolActionRequests) .set({ status: "approved", @@ -10830,7 +10833,7 @@ export function createToolGatewayService( eq(toolActionRequests.status, "pending"), ), ) - .returning(); + .returning()); if (!approved) { throw new ToolGatewayHttpError( 409, diff --git a/server/src/services/tool-review-expiry-work.ts b/server/src/services/tool-review-expiry-work.ts new file mode 100644 index 00000000000..8b7a8caad13 --- /dev/null +++ b/server/src/services/tool-review-expiry-work.ts @@ -0,0 +1,25 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; +import type { createToolGatewayService } from "./tool-gateway.js"; + +export function registerToolReviewExpiryWork( + coordinator: ReturnType, + gateway: Pick, "sweepExpiredActionReviewsForCompany" | "actionReviewExpiryDeadlines">, + canRun: () => boolean, +) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.toolReviewExpiry, + batchSize: 1, + minimumPassIntervalMs: 0, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: (entry, companyId?: string) => companyId === undefined || entry.companyId === companyId, + load: async (companyId?: string) => (await gateway.actionReviewExpiryDeadlines(companyId)) + .map(row => ({ ...row, id: row.companyId })), + check: async entry => { + await gateway.sweepExpiredActionReviewsForCompany(entry.companyId); + const [next] = await gateway.actionReviewExpiryDeadlines(entry.companyId); + return next ? { ...next, id: next.companyId } : null; + }, + }, canRun); +} diff --git a/server/src/services/tool-review-expiry.ts b/server/src/services/tool-review-expiry.ts new file mode 100644 index 00000000000..8db647ce149 --- /dev/null +++ b/server/src/services/tool-review-expiry.ts @@ -0,0 +1,65 @@ +import { and, asc, eq, gt, isNotNull, lte, sql } from "drizzle-orm"; +import { toolActionRequests, toolInvocations, type Db } from "@paperclipai/db"; +import { writeToolReviewChange } from "./tool-review-work-notifications.js"; + +export function toolReviewExpiryService(db: Db, onExpired: (input: { + actionRequestId: string; + invocationId: string; + status: "expired"; + errorCode: string; + errorMessage: string; +}) => Promise) { + return { + async next(): Promise { + const [row] = await db.select({ at: sql`min(${toolActionRequests.expiresAt})` }) + .from(toolActionRequests).where(eq(toolActionRequests.status, "pending")); + return row?.at == null ? null : new Date(row.at).getTime(); + }, + sweep, + sweepForCompany: (companyId: string) => sweep(new Date(), companyId, 1), + async deadlines(companyId?: string) { + const pendingExpiry = and(eq(toolActionRequests.status, "pending"), isNotNull(toolActionRequests.expiresAt)); + const rows = companyId === undefined + ? await db.select({ companyId: toolActionRequests.companyId, + at: sql`min(${toolActionRequests.expiresAt})` }) + .from(toolActionRequests).where(pendingExpiry).groupBy(toolActionRequests.companyId) + : await db.select({ companyId: toolActionRequests.companyId, + at: sql`${toolActionRequests.expiresAt}` }) + .from(toolActionRequests).where(and(pendingExpiry, eq(toolActionRequests.companyId, companyId))) + .orderBy(asc(toolActionRequests.expiresAt)).limit(1); + return rows.map(row => ({ companyId: row.companyId, dueAt: new Date(row.at).getTime() })); + }, + }; + async function sweep(now = new Date(), companyId?: string, maxPages = Infinity) { + let cursor: string | undefined; + let scanned = 0; + for (let page = 0; page < maxPages; page++) { + const rows = await db.select().from(toolActionRequests).where(and( + eq(toolActionRequests.status, "pending"), lte(toolActionRequests.expiresAt, now), + cursor ? gt(toolActionRequests.id, cursor) : undefined, + companyId === undefined ? undefined : eq(toolActionRequests.companyId, companyId), + )).orderBy(asc(toolActionRequests.id)).limit(100); + for (const row of rows) { + const errorCode = "action_expired"; + const errorMessage = "The approval request expired before a decision."; + const changed = await writeToolReviewChange(db, row.companyId, async tx => { + const [updated] = await tx.update(toolActionRequests) + .set({ status: "expired", resolvedAt: now, updatedAt: now }).where(and( + eq(toolActionRequests.id, row.id), eq(toolActionRequests.status, "pending"), + eq(toolActionRequests.updatedAt, row.updatedAt), + )).returning({ id: toolActionRequests.id }); + if (!updated) return false; + await tx.update(toolInvocations).set({ status: "failed", errorCode, errorMessage, + completedAt: now, updatedAt: now }).where(eq(toolInvocations.id, row.invocationId)); + return true; + }); + if (changed) await onExpired({ actionRequestId: row.id, invocationId: row.invocationId, + status: "expired", errorCode, errorMessage }); + } + scanned += rows.length; + if (rows.length < 100) break; + cursor = rows[rows.length - 1].id; + } + return { scanned }; + } +} diff --git a/server/src/services/tool-review-work-notifications.ts b/server/src/services/tool-review-work-notifications.ts new file mode 100644 index 00000000000..19c8e433d75 --- /dev/null +++ b/server/src/services/tool-review-work-notifications.ts @@ -0,0 +1,13 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export function notifyToolReviewChange(transaction: object, companyId: string) { + return notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry, companyId); +} + +export function writeToolReviewChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise { + return db.transaction(async transaction => { + await notifyToolReviewChange(transaction, companyId); + return write(transaction as unknown as Db); + }); +}