mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 22:24:16 +01:00
fix(server): schedule pending tool-review expiry from committed changes (#17453)
## 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
bf3ef12f3a
commit
d108aeeaaf
21 files changed
+905
-47
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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(
|
||||
|
||||
@@ -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';
|
||||
@@ -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';
|
||||
+70
-2
@@ -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",
|
||||
+92
-2
@@ -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",
|
||||
+14
@@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -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),
|
||||
],
|
||||
|
||||
@@ -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),
|
||||
},
|
||||
|
||||
@@ -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<Parameters<typeof createToolGatewayService>[1]>;
|
||||
@@ -203,13 +210,22 @@ function runSessionInput(company: IdRef, agent: IdRef, run: IdRef): Parameters<R
|
||||
describeEmbeddedPostgres("tool gateway service", () => {
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let stopExpiryWorker: (() => Promise<void>) | 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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let gateway: ReturnType<typeof createToolGatewayService>;
|
||||
let fixture: Awaited<ReturnType<typeof createRunFixture>>;
|
||||
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<typeof db.$client.unsafe>[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<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(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<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(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<ReturnType<typeof request>>;
|
||||
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));
|
||||
|
||||
+11
-2
@@ -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) {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
gateway: Pick<ReturnType<typeof createToolGatewayService>, "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);
|
||||
}
|
||||
@@ -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<void>) {
|
||||
return {
|
||||
async next(): Promise<number | null> {
|
||||
const [row] = await db.select({ at: sql<string | Date | null>`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<string | Date>`min(${toolActionRequests.expiresAt})` })
|
||||
.from(toolActionRequests).where(pendingExpiry).groupBy(toolActionRequests.companyId)
|
||||
: await db.select({ companyId: toolActionRequests.companyId,
|
||||
at: sql<Date>`${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 };
|
||||
}
|
||||
}
|
||||
@@ -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<T>(db: Db, companyId: string, write: (transaction: Db) => Promise<T>): Promise<T> {
|
||||
return db.transaction(async transaction => {
|
||||
await notifyToolReviewChange(transaction, companyId);
|
||||
return write(transaction as unknown as Db);
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user