From f6bc653f7f745adaafa26393852fb2f5a99256a4 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 16:33:30 -0500 Subject: [PATCH] fix(server): schedule interrupted tool claims by their deadlines (#17458) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip governs tool execution and keeps durable approval records. > - An interrupted execution may have produced an external effect without a saved final request state. > - The maintenance tick currently scans executing requests even when none exist. > - Every executing claim already has a ten-minute stale deadline. > - This PR schedules that deadline from committed claim changes and clears it after settlement. > - Recovery preserves saved provider outcomes and quarantines unknown outcomes without another provider call. ## Linked Issues or Issue Description Refs #17454. This is the sixth recurring-server sleep slice, stacked on `codex/sleep-05-approved-tool-recovery`. **Problem** The aggregate maintenance tick repeatedly searches for stale executing tool requests. Its settlement can overwrite a saved successful invocation with an unknown-outcome failure after a crash between invocation and request writes. **Proposed Solution** Use the existing review notification boundary and shared scheduler. Read the earliest executing claim deadline through a partial index. At that deadline, lock invocation then request, recheck claim identity, and atomically settle from the saved outcome or quarantine the unknown outcome. **Alternatives Considered** A longer polling interval still queries empty work. A new queue would duplicate existing records. Replaying an interrupted call can duplicate an external effect. Treating every stale request as unknown discards already saved outcomes. ## What Changed - Coalesce sequential review notifications for up to 30 seconds while preserving already loaded recovery deadlines. Do not schedule scans when no notification or execution deadline remains. - Extract interrupted-execution deadline lookup and settlement into the owning tool-review service. - Register its worker at startup. Use committed claim/result changes to update the existing ten-minute deadline; retry failed settlement after 30 seconds. - Preserve terminal invocation results, reasons, and completion times when repairing request state. - Settle uncertain request/invocation state atomically with the existing unknown-outcome reason, without dispatching a provider call. - Notify all gateway claim and settlement paths through the existing review transaction helper. Keep framework wiring and the shared scheduler free of provider-specific behavior. - Add an idempotent concurrent partial index for executing deadlines. Preserve the combined manual sweep and scheduler-disabled startup pass. - Remove interrupted scanning from the aggregate tick. Interaction projection repair remains the next separate duty. ## Verification Current head: `648d8a9a64ded15deeeb331545b92a12d69e277d`. The final commit changes no files and requests CI after the master retarget. A preceding run appeared while that trigger was being published; it was observational delay, not an established CI outage. No test rerun or bypass was used to turn a failure green. Source remains identical to the 198-test passing integration, apart from the already recorded ledger update. Fresh checks and both automated reviews must qualify this current head before merge. Current master-targeted head: `d55e44b376555e4926bb2c2a0e4dcfb431324162`. Parent #17454 landed as `77295faebc8`; a normal merge reconciles that squash. Every conflicted master file was checked against the already-integrated parent, and all matched byte-for-byte. Each interruption-only hunk was retained; the complete integrated source tree matched the 198-test passing `848f1b87b3` exactly before the sole final documentation update. All master migration journal entries are unchanged, with only this PR's executing-deadline index appended. Server types pass. Fresh CI and current-head Greptile/Ponytail are required before the next authorized merge. Latest published integration: `848f1b87b3a6646ae4e56e8ab142f64889d7a69c`, normally merging approved-recovery parent `23868c1751`. All 198 gateway/interruption/startup/deadline cases and server production typechecking pass. Both parent approval fixes and all interruption tests are retained. The two merge conflicts preserved the interruption import alongside the parent's claim-bound hash helper and preserved the parameterized historical test helpers. This remains stacked on #17454; fresh CI and both automated review gates are required after integration. - Review documentation correction at `c985eb9d1b48a481909de6a1c2d89d7af11bfc03`: the focused real-PostgreSQL interruption-worker selection passes all 16 cases; 124 unrelated cases were filtered out. The docs now correctly name 16 cases and six saved terminal statuses. No production behavior changed in the correction. Fresh remote CI/review is pending after integrating the verified parent. Approved delegated Test-tab calls now evaluate caller and target policies in the existing claim transaction. Real HTTP regressions first showed a missing caller charge and shared-limit self-rejection. All 78 connection-management cases and 92 gateway cases now pass on the owning branch, along with server production typechecking. All 78 connection-management cases also pass at combined tip `d25f830db89c1623bff3547bfc68015c1c66e0cf`. The three new cases cover shared charges, separate charges, rollback on denial, duplicate approval, and rejection of a later authorized call. Intermediate counter/identity fixture failures and a reused-database Slack collision are recorded in [the owning audit](https://github.com/paperclipai/paperclip/blob/f6a5949fb726f71c247950714adecb275655284f/doc/idle-approved-tool-recovery.md#delegated-test-tab-admission-after-policy-changes). The earlier combined checks below remain evidence for their stated revisions. Current-head CI and review remain required. Current head: `f6a5949fb726f71c247950714adecb275655284f`. This remains a single-duty PR stacked on #17454. - 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/eaac8d343e5ef4d2c13cdf7cb7607cb5a498e9a2/doc/plans/2026-10-24-recurring-server-sleep-stack.md). - Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355 startup/cancellation/recovery/dispatch/preservation tests**, **75 connection-management tests**, and production server typechecking pass. These combined checks do not mean every intermediate PR independently ran the full repository suite. - Real master-to-stack database upgrade passes: **379 unchanged prior migration receipts plus 25 new migrations**, with the gateway responsibility fixture preserved. - Owning-branch checks: 84 runner installation/settlement cases plus runner typechecking and generated TypeScript/Rust profile agreement; 253 GitHub review workflow cases. Stop-attribution and held-worker readiness regressions failed before their fixes. - Failure accounting: the initial combined run passed 349 cases and failed six idle assertions in a reused preservation database containing old cancelled runs. Repeating the combined command with a fresh preservation database passed. Earlier invalid project selection, merge import/type errors, and the interrupted full local suite are not counted as passing checks. - **Latest-head CI and review remain required.** Prior CI findings include GitHub idle observation, route teardown, attention cleanup, touch-picker E2E, workspace health, and voice-cleanup timing; local passes do not erase those failures. No full local repository-suite, deployed sleep, fleet cost, or live Copilot qualification claim is made. ## Risks - Signals cover the existing single-active-writer database owner. Independent writers and replicas need an explicit wake contract. - The ten-minute policy is unchanged. A provider call that outlives its claim remains an uncertain outcome; this worker never retries the external call. - Interaction projection follows authoritative settlement outside its transaction. Its separate existing repair duty still handles an interrupted projection. - Non-executing review writes can cause one empty reconciliation. Without an executing claim or failed attempt, no recurring deadline remains. - Cloud eligibility and external-alarm coverage remain conservative. Worker quiet does not prove actual host or database suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/idle-interrupted-tool-recovery.md | 116 ++++++++ ...2026-10-24-recurring-server-sleep-stack.md | 52 +++- packages/db/src/client.ts | 1 + .../src/migrations/0385_open_madame_web.sql | 1 + ...{0380_snapshot.json => 0385_snapshot.json} | 101 ++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/tool_access.ts | 1 + .../server-startup-feedback-export.test.ts | 2 + .../__tests__/tool-gateway-service.test.ts | 253 ++++++++++++++++++ server/src/index.ts | 11 +- .../services/delivery-work-notifications.ts | 1 + server/src/services/tool-gateway.ts | 90 ++----- .../services/tool-review-interruption-work.ts | 16 ++ .../src/services/tool-review-interruption.ts | 61 +++++ .../tool-review-work-notifications.ts | 1 + 15 files changed, 643 insertions(+), 71 deletions(-) create mode 100644 doc/idle-interrupted-tool-recovery.md create mode 100644 packages/db/src/migrations/0385_open_madame_web.sql rename packages/db/src/migrations/meta/{0380_snapshot.json => 0385_snapshot.json} (99%) create mode 100644 server/src/services/tool-review-interruption-work.ts create mode 100644 server/src/services/tool-review-interruption.ts diff --git a/doc/idle-interrupted-tool-recovery.md b/doc/idle-interrupted-tool-recovery.md new file mode 100644 index 00000000000..c02ecdd97ec --- /dev/null +++ b/doc/idle-interrupted-tool-recovery.md @@ -0,0 +1,116 @@ +# Interrupted tool-action recovery + +An executing tool-action request has an existing ten-minute stale deadline from +its `updatedAt` claim timestamp. Recovery now registers the earliest actual +claim deadline with the shared scheduler. It no longer scans executing requests +on every aggregate maintenance tick. + +## Ownership and producers + +The gateway owns execution claims, provider calls, and result settlement. The +new interruption service owns only the deadline query and stale-row settlement. +The worker uses the existing deadline registration and exposes no provider +protocol or signed argument representation to the scheduler. + +The review write helper signals interruption alongside approved recovery and +pending expiry. All three agent claim paths and the Test-tab claim use that +helper. Test-tab outcomes already commit both rows inside it. Agent pre-dispatch +and execution failure settlement uses the same notification transaction, and +successful or failed direct execution request updates now notify before writing. +Administrative review transitions continue to use the same existing boundary. + +Unrelated pending review changes can cause a bounded empty reconciliation. They +do not create a periodic deadline. Process-local signals cover the existing +single-active-writer database owner; independent writers require a wake contract. + +## Settlement and quiet states + +The query uses a partial executing-request index and reads one earliest row. +There is no deadline after all executing claims settle. Retained terminal +requests and configured-but-idle tools produce no recurring worker queries. + +At the deadline, settlement locks invocation then request, matching the existing +execution-failure lock order. It rechecks the claim status and timestamp after +locking. A changed or completed claim cannot be settled from an earlier scan. +Unknown outcomes atomically mark both rows failed with +`tool_execution_outcome_unknown`. Recovery never calls the external provider. + +A saved terminal invocation is authoritative. Recovery repairs its request to +executed or failed without replacing the saved result, failure reason, or +completion time with an unknown-outcome failure. This covers a crash between +legacy invocation completion and request settlement. Interaction projection +runs after the transaction through the existing gateway owner. A failed +projection still relies on the separate projection-repair duty. + +A database failure rolls back both outcome rows and retains a 30-second retry. +Startup reconstructs saved deadlines. Drain pauses dispatch without forgetting +the first claim or failed retry; resume processes overdue claims. Completion +notifications cancel unnecessary future scans. The scheduler-disabled path keeps +one startup pass. The public combined review sweep retains its existing duties. + +## Verification and limits + +Sixteen new real-PostgreSQL cases cover real foreground execution activation +and completion, configured idle and terminal history, restart/deadline timing, +six saved terminal invocation states, first-write drain/resume, outer commit and +rollback, transactional failure/retry, concurrent claim change, and multi-page +settlement. The service/startup selection passes 136 tests on the combined parent-review +head. Workspace typecheck/build and migration safety checks pass on that head. +Migration application and reapplication passed. The combined run also exercised +469 tool-access tests: 468 passed and an unchanged connection-archive case timed +out under concurrent build load. That case passes in isolation. The separate full tool-access suite passes all +469 tests and gateway acceptance passes all 92 (697 across the final selections). +Fresh full CI/review remains required. + +The first typecheck caught optional approval IDs losing narrowing in transaction +callbacks; capturing them before the callback fixes it. Two new fault/transaction +cases initially waited for idle before advancing their next fake-clock retry. +They now advance time while waiting for the authoritative outcome, and the full +selection passes. These are fixture corrections, not permission to clear an +unresolved work hold. + +The ten-minute policy is unchanged. Active provider execution, unknown external +outcomes, and other worker obligations still prevent unsafe sleep. No Cloud +eligibility or external-alarm coverage changes here. Worker quiet is not evidence +of whole-instance or database suspension. + +## Sequential-write cost correction (October 27) + +Review found that sequential tenant writes could each request a global +interruption pass, even when no execution was stale. A real-SQL regression +created ten tool review requests through the gateway and allowed each preceding +pass to finish. It measured 11 stale-claim scans including startup. Concurrent +notification coalescing alone did not cover this sequence. + +The worker now uses the existing deadline helper's 30-second notification +coalescing window, matching the former default maintenance cadence. The same +regression measures one startup scan and one follow-up scan for those ten +requests, then no worker SQL through an additional idle hour. A loaded stale +claim deadline only 500 milliseconds away still preempts the window and +quarantines the unknown outcome without invoking the provider. Foreground +execution and result settlement remain immediate. Newly discovered work can +wait up to this window; the ten-minute stale-claim policy is unchanged. + +This is the bounded-frequency alternative raised in review. Discovery is still +instance-wide, and real stale backlogs still drain through the existing pages. +It does not claim company-isolated scans or constant cost for a real backlog. +No notification and no saved execution deadline means no periodic scan. No new +table, index, or scheduler is added. Startup, drain, retry and shutdown retain +the existing owners. + +The first full verification passed 175 cases and failed three fake-clock waits: +settlement writes leave a notification pending for its coalesced follow-up, so +waiting for the idle hold without advancing that armed deadline cannot finish. +An initial shared clock adjustment then advanced an unrelated five-second +expiry test too far (177 passed, one failed). The adjustment is now confined to +the interruption tests. It lets real SQL finish and advances only to an actually +armed short follow-up. The imminent-deadline test checks quarantine before that +30-second window and separately waits for eventual quiet. Product assertions +were retained. + +The final complete gateway-service and startup-wiring suites pass all 178 cases +without skips, and server typechecking passes, on owner source based on +`e255d8eff3363cf0bb63541bdc7ef055d4321f83`. The attempted third test selector +`deadline-work.test.ts` did not exist at this owner; no separate helper-suite +pass is claimed. The shared helper is exercised by the full gateway suites. +Fresh combined-source checks and remote review remain required. 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 b699f1b242d..b8976fb34fb 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; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), historical argument identity repair verified and awaiting fresh CI/review; other review duties pending | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), merged after current-head CI and both 5/5 reviews; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), integration verified and awaiting fresh CI/review; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), interaction-link index verified and awaiting fresh CI/review | | 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 | @@ -230,3 +230,53 @@ tool-access service tests pass after enabling the existing external database path independently of embedded support. Its initial browser shard failed during unrelated plan-task fixture company deletion with a concurrent activity-log FK violation. Inspect fresh-head CI before deciding whether that needs a repair. + + +## Slice 06: interrupted tool executions + +Branch `codex/sleep-06-interrupted-tool-recovery` is based on #17454 and includes +its one-shot outcome-write repair. The next deadline is the minimum executing +claim timestamp plus the existing ten minutes. The new service locks invocation +then request, rechecks claim identity, preserves saved terminal outcomes, and +atomically quarantines unknown outcomes without calling a provider. All claim +producers and request settlements use the existing review notification boundary. +A partial index supports the first deadline. The aggregate tick now excludes +interrupted claims; projection repair remains its own pending slice. + +Sixteen new worker cases and the existing service/startup selection pass +(136 total on the combined head). Workspace typecheck/build and migration checks +pass. Two fake-clock fixture ordering failures were corrected to wait for real +settlement while advancing retry time. The combined tool-access run had one +unchanged connection-archive timeout; it passed in isolation. The full suite +and gateway acceptance are running separately before readiness is claimed. +No PR is claimed ready until fresh CI and review pass. + +## Next owner audit: tool-review interaction projection + +The remaining aggregate tool-review duty selects terminal requests whose +interaction result does not match the authoritative request state. Recovery +currently reads up to 100 joined rows. Move this duty to its own transaction- +notified worker, with retries only while projection debt remains. Do not add +an outbox table merely to migrate the scan. + +Audit the final two direct request writers in the gateway as well as existing +review helpers, expiry, interruption, and Test-tab outcome settlement. Preserve +the existing reflection owner: it locks issue/interaction/request, checks the +current request state before writing the card, then publishes activity and +requests continuation. A no-op reflection can mean an ownership or lifecycle +condition, not successful reconciliation. Retained terminal requests with +already matching cards must be quiet. Unprojectable/mismatched durable state +must not be silently dropped. Paginate without starving later candidates. + +Continuation delivery already has its own coordinator. A projection write is +not proof that downstream delivery or activity publication completed. Identify +the exact existing recovery owner before removing any backstop, and keep +provider/private payload interpretation out of the scheduler. + +Final slice-06 local verification: the isolated tool-access rerun passes all +469 tests, gateway acceptance passes 92, and the combined service/startup +selection passed 136. All 697 selected tests pass on the combined head. The +initial concurrent-run timeout is retained above as evidence, not hidden. +Workspace typecheck/build and migration safety/numbering pass. The first four +PRs are mergeable and have all checks passed at their recorded heads. #17454 +is mergeable with fresh checks and review in progress at `7a1cf03e62`. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index b3d66e14855..3e4cce15e2e 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -503,6 +503,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0369_late_logan.sql", "0381_mean_scream.sql", "0382_narrow_the_executioner.sql", + "0385_open_madame_web.sql", "0384_funny_adam_warlock.sql", "0383_tool_review_company_expiry.sql", ]); diff --git a/packages/db/src/migrations/0385_open_madame_web.sql b/packages/db/src/migrations/0385_open_madame_web.sql new file mode 100644 index 00000000000..19ce52b5cdb --- /dev/null +++ b/packages/db/src/migrations/0385_open_madame_web.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_action_requests_executing_deadline_idx" ON "tool_action_requests" USING btree ("updated_at","id") WHERE "tool_action_requests"."status" = 'executing'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0380_snapshot.json b/packages/db/src/migrations/meta/0385_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0380_snapshot.json rename to packages/db/src/migrations/meta/0385_snapshot.json index 5e635e8be52..5564d696be5 100644 --- a/packages/db/src/migrations/meta/0380_snapshot.json +++ b/packages/db/src/migrations/meta/0385_snapshot.json @@ -1,6 +1,6 @@ { - "id": "bfda87fa-662b-47bb-87c0-dabc814aebd7", - "prevId": "2a9439e3-4a2b-4a20-ab2b-2aee3db8cb03", + "id": "9965d300-e355-4f47-ba0a-951ed3b4b8a2", + "prevId": "4f96f70a-fd19-4d35-ad8a-8d5559b83cbd", "version": "7", "dialect": "postgresql", "tables": { @@ -52812,6 +52812,88 @@ "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_executing_deadline_idx": { + "name": "tool_action_requests_executing_deadline_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'executing'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_approved_idx": { + "name": "tool_action_requests_approved_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'approved'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -55222,6 +55304,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": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 6b79f74c167..f4cfed69f11 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2682,6 +2682,13 @@ "when": 1793060791728, "tag": "0384_funny_adam_warlock", "breakpoints": true + }, + { + "idx": 385, + "version": "7", + "when": 1793060791729, + "tag": "0385_open_madame_web", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/tool_access.ts b/packages/db/src/schema/tool_access.ts index f1e788626e3..9f79743d6c3 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -836,6 +836,7 @@ export const toolActionRequests = pgTable( .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_executing_deadline_idx").on(table.updatedAt, table.id).where(sql`${table.status} = 'executing'`), index("tool_action_requests_approved_idx").on(table.id).where(sql`${table.status} = 'approved'`), 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 658f5c418ff..1a9fcf51576 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -69,6 +69,8 @@ const { stop: vi.fn(async () => {}), }, toolGateway: { + settleInterruptedActionReviews: vi.fn(async () => ({ scanned: 0 })), + nextActionReviewInterruptionAt: vi.fn(async () => null), recoverApprovedActionReviews: vi.fn(async () => ({ scanned: 0, retry: false })), approvedActionReviewCompanies: vi.fn(async () => []), sweepActionReviews: vi.fn(async () => ({ scanned: 0 })), diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index 4a014b22ab0..35dabccbe22 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -1,3 +1,5 @@ +import { registerToolReviewInterruptionWork } from "../services/tool-review-interruption-work.js"; +import { writeToolReviewChange } from "../services/tool-review-work-notifications.js"; import { deleteCompany } from "../services/company-deletion.js"; import { subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; @@ -318,6 +320,257 @@ describeEmbeddedPostgres("tool gateway service", () => { expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); } + async function testRequest(provider: NonNullable, rateLimited = false) { + await db.insert(companyMemberships).values({ companyId: fixture.company.id, principalType: "user", + principalId: "board", status: "active", membershipRole: "owner" }); + const remote = await createRemoteMcpToolFixture(db, fixture.company.id); + await db.update(toolCatalogEntries).set({ isReadOnly: false, riskLevel: "write" }) + .where(eq(toolCatalogEntries.id, remote.catalogEntry.id)); + await db.insert(toolPolicies).values({ companyId: fixture.company.id, name: "Review connected writes", + policyType: "require_approval", priority: 10, selectors: { connectionId: remote.connection.id } }); + if (rateLimited) await db.insert(toolPolicies).values({ companyId: fixture.company.id, name: "One reviewed test call", + policyType: "rate_limit", priority: 1, selectors: { connectionId: remote.connection.id }, + config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } }); + gateway = createTestToolGatewayService(db, { remoteHttpRequest: provider }); + const result = await gateway.executeTestCall({ companyId: fixture.company.id, connectionId: remote.connection.id, + agentId: fixture.agent.id, userId: "board", toolName: "needs_input", parameters: { value: "reviewed" } }); + expect(result.decision).toBe("ask_first"); + const [pending] = await db.select().from(toolActionRequests); + return { pending, remote }; + } + function providerFixture() { + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const provider = vi.fn(async (_url, init) => { + const payload = JSON.parse(String(init.body)); + expect(payload.method).toBe("tools/call"); + entered(); await gate; + return Response.json({ jsonrpc: "2.0", id: payload.id, + result: { content: [{ type: "text", text: "Saved once" }] } }); + }); + return { provider, release, started }; + } + + describe("interrupted review recovery worker", () => { + async function startInterruption() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerToolReviewInterruptionWork(coordinator, gateway, () => !draining).ready; + } + async function settle(ms = 60_000) { + await vi.advanceTimersByTimeAsync(ms); + if (!draining) await vi.waitFor(async () => { + // Finish real SQL before following an armed coalescing/retry deadline. + // Do not jump across a future business deadline while waiting for I/O. + if (idleWorkSnapshot().active > 0) { + await db.execute(sql`select pg_sleep(0.01)`); + const next = coordinator!.nextWakeAt(); + if (next !== null && next <= Date.now() + 30_000) { + await vi.advanceTimersByTimeAsync(Math.max(0, next - Date.now())); + } + } + 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 savedClaim(source = db, age = 0) { + const pending = await request(); + const claimedAt = new Date(Date.now() - age); + await writeToolReviewChange(source, fixture.company.id, async tx => { + await tx.update(toolInvocations).set({ status: "executing", startedAt: claimedAt, updatedAt: claimedAt }) + .where(eq(toolInvocations.id, pending.invocationId)); + await tx.update(toolActionRequests).set({ status: "executing", updatedAt: claimedAt }) + .where(eq(toolActionRequests.id, pending.id)); + }); + return { ...pending, updatedAt: claimedAt }; + } + async function unknown(pending: Awaited>) { + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("failed"); + expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0]) + .toMatchObject({ status: "failed", errorCode: "tool_execution_outcome_unknown" }); + } + + it("coalesces sequential review writes into one interruption scan per window", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + const previous = db.$client.options.debug; + const scans: string[] = []; + db.$client.options.debug = (_connection, query) => { + if (query.includes('from "tool_action_requests"') && query.includes('"updated_at" <=')) scans.push(query); + }; + try { + await startInterruption(); + const firstPassAt = Date.now(); + expect(scans).toHaveLength(1); + for (let i = 0; i < 10; i++) { + await request(); + await vi.advanceTimersByTimeAsync(1); + // Let each previous pass finish; overlapping calls already coalesce. + await db.execute(sql`select pg_sleep(0.01)`); + } + expect(scans).toHaveLength(1); + expect(coordinator!.nextWakeAt()).toBe(firstPassAt + 30_000); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => { + expect(scans).toHaveLength(2); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }); + await quiet(); + expect(scans).toHaveLength(2); + } finally { db.$client.options.debug = previous; } + }); + + it("keeps a loaded stale-claim deadline ahead of review notification coalescing", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + const pending = await savedClaim(db, 10 * 60_000 - 500); + const deadline = pending.updatedAt.getTime() + 10 * 60_000; + await startInterruption(); + expect(coordinator!.nextWakeAt()).toBe(deadline); + await request(); + await vi.advanceTimersByTimeAsync(1); + await db.execute(sql`select pg_sleep(0.01)`); + expect(coordinator!.nextWakeAt()).toBe(deadline); + await vi.advanceTimersByTimeAsync(499); + await vi.waitFor(async () => { await unknown(pending); }); + expect(Date.now()).toBeLessThan(deadline + 1000); + await quiet(); + }); + + it("arms from a real execution claim and disarms after the provider result", async () => { + await startInterruption(); await quiet(); + const remote = providerFixture(); + const { pending } = await testRequest(remote.provider); + const running = gateway.approveActionRequest({ companyId: fixture.company.id, + actionRequestId: pending.id, actor: { userId: "board" } }); + await Promise.race([remote.started, running]); + try { + await settle(); + const [claim] = await db.select().from(toolActionRequests); + expect(claim.status).toBe("executing"); + expect(coordinator!.nextWakeAt()).toBe(claim.updatedAt.getTime() + 10 * 60_000); + const reads = vi.spyOn(db, "select"); + await settle(60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + } finally { remote.release(); } + await running; await quiet(); + expect((await db.select().from(toolActionRequests))[0].status).toBe("executed"); + expect(remote.provider).toHaveBeenCalledTimes(1); + }); + + it("reconstructs the saved deadline and quarantines only when due", async () => { + const pending = await savedClaim(); await startInterruption(); + expect(coordinator!.nextWakeAt()).toBe(pending.updatedAt.getTime() + 10 * 60_000); + await settle(9 * 60_000); + expect((await db.select().from(toolActionRequests))[0].status).toBe("executing"); + await settle(2 * 60_000); await unknown(pending); await quiet(); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); + }); + + it.each(["succeeded", "failed", "denied", "timed_out", "rate_limited", "cancelled"] as const)("preserves a saved %s invocation while repairing the request", async status => { + const pending = await savedClaim(db, 11 * 60_000); + const completedAt = new Date(Date.now() - 5 * 60_000); + await db.update(toolInvocations).set({ status, completedAt, resultSummary: { summary: "Saved provider result" }, + errorCode: status === "succeeded" ? null : "saved_provider_failure" }).where(eq(toolInvocations.id, pending.invocationId)); + await startInterruption(); await quiet(); + expect((await db.select().from(toolActionRequests))[0]).toMatchObject({ + status: status === "succeeded" ? "executed" : "failed", resolvedAt: completedAt }); + expect((await db.select().from(toolInvocations))[0]).toMatchObject({ status, completedAt, + errorCode: status === "succeeded" ? null : "saved_provider_failure", resultSummary: { summary: "Saved provider result" } }); + }); + + it("retains a first claim committed during drain without passive database scans", async () => { + await startInterruption(); await quiet(); draining = true; + const pending = await savedClaim(); await settle(); + const reads = vi.spyOn(db, "select"); + await settle(60 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; await settle(); await unknown(pending); await quiet(); + }); + + it.each([false, true])("waits for outer execution commit (rollback %s)", async rollback => { + const pending = await request(); await startInterruption(); await quiet(); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const aborted = new Error("fixture rollback"); + const running = db.transaction(async tx => { + await writeToolReviewChange(tx as unknown as typeof db, fixture.company.id, async nested => { + await nested.update(toolActionRequests).set({ status: "executing", updatedAt: new Date() }) + .where(eq(toolActionRequests.id, pending.id)); + }); + entered(); await gate; if (rollback) throw aborted; + }).catch(error => { if (error !== aborted) throw error; }); + await ready; + try { + await vi.advanceTimersByTimeAsync(11 * 60_000); + expect((await db.select().from(toolActionRequests))[0].status).toBe("pending"); + } finally { release(); await running; } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + if (rollback) expect((await db.select().from(toolActionRequests))[0].status).toBe("pending"); + else await unknown(pending); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + await quiet(); + }); + + it("rolls back both outcome rows on failure and retries after drain", async () => { + const pending = await savedClaim(db, 11 * 60_000); + await db.execute(sql`create function sleep_review_fail() returns trigger language plpgsql as $$ + begin raise exception 'interruption fixture failure'; end $$`); + await db.execute(sql`create trigger sleep_review_fail before update on tool_invocations + for each row when (new.error_code = 'tool_execution_outcome_unknown') execute function sleep_review_fail()`); + await startInterruption(); expect(errors.length).toBeGreaterThan(0); errors.length = 0; + expect((await db.select().from(toolActionRequests))[0].status).toBe("executing"); + expect((await db.select().from(toolInvocations))[0].status).toBe("executing"); + draining = true; + await db.execute(sql`drop trigger sleep_review_fail on tool_invocations`); + await settle(60 * 60_000); + draining = false; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + await unknown(pending); + }, { timeout: 5000 }); + await quiet(); + }); + + it("rechecks a changed claim after selecting a stale candidate", async () => { + const pending = await savedClaim(db, 11 * 60_000); + const transaction = db.transaction.bind(db); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const intercepted = vi.spyOn(db, "transaction").mockImplementationOnce(async callback => { + entered(); await gate; return transaction(callback); + }); + const sweep = gateway.settleInterruptedActionReviews(); + await ready; + const updatedAt = new Date(); + try { + await writeToolReviewChange(db, fixture.company.id, tx => tx.update(toolActionRequests).set({ updatedAt }) + .where(eq(toolActionRequests.id, pending.id))); + } finally { release(); } + await sweep; intercepted.mockRestore(); + expect((await db.select().from(toolActionRequests))[0].status).toBe("executing"); + await startInterruption(); + expect(coordinator!.nextWakeAt()).toBe(updatedAt.getTime() + 10 * 60_000); + }); + + it("settles more than one page of interrupted claims", async () => { + const pending = await savedClaim(db, 11 * 60_000); + await db.insert(toolActionRequests).values(Array.from({ length: 101 }, () => ({ ...pending, + id: randomUUID(), interactionId: null, status: "executing" as const }))); + await startInterruption(); await quiet(); + const rows = await db.select().from(toolActionRequests); + expect(rows).toHaveLength(102); expect(rows.every(row => row.status === "failed")).toBe(true); + }); + }); + describe("approved review recovery worker", () => { async function startRecovery() { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, diff --git a/server/src/index.ts b/server/src/index.ts index e9fc24fcf0b..50627307a54 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js"; import { registerToolReviewRecoveryWork } from "./services/tool-review-recovery-work.js"; import { registerToolReviewExpiryWork } from "./services/tool-review-expiry-work.js"; import { registerGatewayCredentialExpiryWork } from "./services/gateway-credential-expiry-work.js"; @@ -1479,7 +1480,13 @@ async function startServerWithDatabaseTeardown( } else { await app.locals.toolGateway.recoverApprovedActionReviews(); } - return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false }); + if (config.heartbeatSchedulerEnabled) { + await registerToolReviewInterruptionWork(deliveryWork, app.locals.toolGateway, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + } else { + await app.locals.toolGateway.settleInterruptedActionReviews(); + } + return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false, includeInterrupted: false }); }).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed")); await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled ? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway, @@ -1910,7 +1917,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "periodic tool connection health sweep failed"); })); - trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false }).catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); + trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false, includeInterrupted: 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/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 923ae4b7979..9c943677dac 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", + toolReviewInterruption: "tool-review-interruption", toolReviewRecovery: "tool-review-recovery", toolReviewExpiry: "tool-review-expiry", gatewayCredentialExpiry: "gateway-credential-expiry", diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 57763f8b8d5..2d94a361de2 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,7 +1,8 @@ +import { toolReviewInterruptionService } from "./tool-review-interruption.js"; import { bindVerifiedToolApprovalArguments, claimToolTestApproval, releaseToolApproval, persistToolTestOutcome, startToolInvocation, type ToolApprovalClaim } from "./tool-approval-execution.js"; import { approvedToolReviewCompanies, prepareApprovedToolReview, recoverApprovedToolReviews } from "./tool-review-recovery.js"; import { toolReviewExpiryService } from "./tool-review-expiry.js"; -import { writeToolReviewChange } from "./tool-review-work-notifications.js"; +import { notifyToolReviewChange, 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"; @@ -46,7 +47,6 @@ import { asc, desc, eq, - gt, inArray, isNull, lte, @@ -7987,6 +7987,7 @@ export function createToolGatewayService( const [actionRequest] = await tx .select({ + companyId: toolActionRequests.companyId, status: toolActionRequests.status, updatedAt: toolActionRequests.updatedAt, }) @@ -8001,6 +8002,7 @@ export function createToolGatewayService( return false; } + await notifyToolReviewChange(tx, actionRequest.companyId); await tx .update(toolInvocations) .set({ @@ -8476,10 +8478,10 @@ export function createToolGatewayService( }) .where(eq(toolInvocations.id, invocation.id)); void emitConnectionInvoked(db, invocation.id); - await db + await writeToolReviewChange(db, claimed.companyId, tx => tx .update(toolActionRequests) .set({ status: "executed", resolvedAt: now, updatedAt: now }) - .where(eq(toolActionRequests.id, claimed.id)); + .where(eq(toolActionRequests.id, claimed.id))); await reflectToolActionInteractionLifecycle({ actionRequestId: claimed.id, status: "executed", @@ -8818,6 +8820,11 @@ export function createToolGatewayService( }; } + const reviewInterruption = toolReviewInterruptionService(db, async settled => { + void emitConnectionInvoked(db, settled.invocationId); + await reflectToolActionInteractionLifecycle(settled); + }); + const reviewExpiry = toolReviewExpiryService(db, async expired => { void emitConnectionInvoked(db, expired.invocationId); await reflectToolActionInteractionLifecycle(expired); @@ -9910,6 +9917,8 @@ export function createToolGatewayService( return buildTestCallStatus(actionRequest, invocation); }, + settleInterruptedActionReviews: reviewInterruption.sweep, + nextActionReviewInterruptionAt: reviewInterruption.next, sweepExpiredActionReviews: reviewExpiry.sweep, nextActionReviewExpiryAt: reviewExpiry.next, actionReviewExpiryDeadlines: reviewExpiry.deadlines, @@ -9941,7 +9950,7 @@ export function createToolGatewayService( }, companyId); }, - async sweepActionReviews(input: { includeExpiry?: boolean; includeApproved?: boolean } = {}) { + async sweepActionReviews(input: { includeExpiry?: boolean; includeApproved?: boolean; includeInterrupted?: 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 @@ -9976,62 +9985,9 @@ export function createToolGatewayService( errorMessage: invocation.errorMessage, resultSummary: invocation.resultSummary?.summary, }); - const now = new Date(); - const staleAt = new Date(now.getTime() - 10 * 60_000); - let cursor: string | undefined; - let scanned = input.includeExpiry === false ? 0 : (await reviewExpiry.sweep(now)).scanned; + let scanned = input.includeExpiry === false ? 0 : (await reviewExpiry.sweep()).scanned; if (input.includeApproved !== false) scanned += (await this.recoverApprovedActionReviews()).scanned; - for (;;) { - const rows = await db - .select() - .from(toolActionRequests) - .where( - and( - eq(toolActionRequests.status, "executing"), - lte(toolActionRequests.updatedAt, staleAt), - cursor ? gt(toolActionRequests.id, cursor) : undefined, - ), - ) - .orderBy(asc(toolActionRequests.id)) - .limit(100); - for (const row of rows) { - 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 }) - .where( - and( - eq(toolActionRequests.id, row.id), - eq(toolActionRequests.status, row.status), - eq(toolActionRequests.updatedAt, row.updatedAt), - ), - ) - .returning(); - if (!changed) continue; - await db - .update(toolInvocations) - .set({ - status: "failed", - errorCode, - errorMessage, - completedAt: now, - updatedAt: now, - }) - .where(eq(toolInvocations.id, row.invocationId)); - void emitConnectionInvoked(db, row.invocationId); - await reflectToolActionInteractionLifecycle({ - actionRequestId: row.id, - status, - errorCode, - errorMessage, - }); - } - scanned += rows.length; - if (rows.length < 100) break; - cursor = rows[rows.length - 1].id; - } + if (input.includeInterrupted !== false) scanned += (await reviewInterruption.sweep()).scanned; return { scanned }; }, @@ -11251,7 +11207,8 @@ export function createToolGatewayService( }) .where(eq(toolInvocations.id, invocationId)); if (input.approvedActionRequestId) { - const [executedRequest] = await db + const actionRequestId = input.approvedActionRequestId; + const [executedRequest] = await writeToolReviewChange(db, session.companyId, tx => tx .update(toolActionRequests) .set({ status: "executed", @@ -11260,11 +11217,11 @@ export function createToolGatewayService( }) .where( and( - eq(toolActionRequests.id, input.approvedActionRequestId), + eq(toolActionRequests.id, actionRequestId), eq(toolActionRequests.status, "executing"), ), ) - .returning({ id: toolActionRequests.id }); + .returning({ id: toolActionRequests.id })); if (executedRequest) { await reflectToolActionInteractionLifecycle({ actionRequestId: executedRequest.id, @@ -11384,7 +11341,8 @@ export function createToolGatewayService( .where(eq(toolInvocations.id, invocationId)); void emitConnectionInvoked(db, invocationId); if (input.approvedActionRequestId) { - const [failedRequest] = await db + const actionRequestId = input.approvedActionRequestId; + const [failedRequest] = await writeToolReviewChange(db, session.companyId, tx => tx .update(toolActionRequests) .set({ status: "failed", @@ -11393,11 +11351,11 @@ export function createToolGatewayService( }) .where( and( - eq(toolActionRequests.id, input.approvedActionRequestId), + eq(toolActionRequests.id, actionRequestId), eq(toolActionRequests.status, "executing"), ), ) - .returning({ id: toolActionRequests.id }); + .returning({ id: toolActionRequests.id })); if (failedRequest) { await reflectToolActionInteractionLifecycle({ actionRequestId: failedRequest.id, diff --git a/server/src/services/tool-review-interruption-work.ts b/server/src/services/tool-review-interruption-work.ts new file mode 100644 index 00000000000..9053cb1c4dd --- /dev/null +++ b/server/src/services/tool-review-interruption-work.ts @@ -0,0 +1,16 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { createToolGatewayService } from "./tool-gateway.js"; + +export function registerToolReviewInterruptionWork( + coordinator: ReturnType, + gateway: Pick, "settleInterruptedActionReviews" | "nextActionReviewInterruptionAt">, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.toolReviewInterruption, { + retryMs: 30_000, notificationCoalesceMs: 30_000, canRun, + run: () => gateway.settleInterruptedActionReviews(), + next: () => gateway.nextActionReviewInterruptionAt(), + }); +} diff --git a/server/src/services/tool-review-interruption.ts b/server/src/services/tool-review-interruption.ts new file mode 100644 index 00000000000..de8769f3da1 --- /dev/null +++ b/server/src/services/tool-review-interruption.ts @@ -0,0 +1,61 @@ +import { and, asc, eq, gt, lte } from "drizzle-orm"; +import { toolActionRequests, toolInvocations, type Db } from "@paperclipai/db"; +import { writeToolReviewChange } from "./tool-review-work-notifications.js"; + +const STALE_EXECUTION_MS = 10 * 60_000; + +export function toolReviewInterruptionService(db: Db, onSettled: (input: { + actionRequestId: string; + invocationId: string; + status: "executed" | "failed"; + errorCode: string | null; + errorMessage: string | null; + resultSummary?: string; +}) => Promise) { + return { + async next(): Promise { + const [row] = await db.select({ at: toolActionRequests.updatedAt }).from(toolActionRequests) + .where(eq(toolActionRequests.status, "executing")).orderBy(asc(toolActionRequests.updatedAt)).limit(1); + return row ? row.at.getTime() + STALE_EXECUTION_MS : null; + }, + async sweep(now = new Date()) { + const staleAt = new Date(now.getTime() - STALE_EXECUTION_MS); + let cursor: string | undefined; + let scanned = 0; + for (;;) { + const rows = await db.select().from(toolActionRequests).where(and( + eq(toolActionRequests.status, "executing"), lte(toolActionRequests.updatedAt, staleAt), + cursor ? gt(toolActionRequests.id, cursor) : undefined, + )).orderBy(asc(toolActionRequests.id)).limit(100); + for (const row of rows) { + const settled = await writeToolReviewChange(db, row.companyId, async tx => { + const [invocation] = await tx.select().from(toolInvocations) + .where(eq(toolInvocations.id, row.invocationId)).for("update"); + const [request] = await tx.select().from(toolActionRequests) + .where(eq(toolActionRequests.id, row.id)).for("update"); + if (!invocation || request?.status !== "executing" || + request.updatedAt.getTime() !== row.updatedAt.getTime()) return null; + const completed = ["succeeded", "failed", "denied", "timed_out", "rate_limited", "cancelled"].includes(invocation.status); + const status = invocation.status === "succeeded" ? "executed" as const : "failed" as const; + const errorCode = completed ? invocation.errorCode : "tool_execution_outcome_unknown"; + const errorMessage = completed ? invocation.errorMessage + : "Execution was interrupted; the external outcome is unknown. Inspect the provider before retrying."; + if (!completed) { + await tx.update(toolInvocations).set({ status: "failed", errorCode, errorMessage, + completedAt: now, updatedAt: now }).where(eq(toolInvocations.id, row.invocationId)); + } + await tx.update(toolActionRequests).set({ status, resolvedAt: invocation.completedAt ?? now, updatedAt: now }) + .where(eq(toolActionRequests.id, row.id)); + return { actionRequestId: row.id, invocationId: row.invocationId, status, + errorCode, errorMessage, resultSummary: invocation.resultSummary?.summary }; + }); + if (settled) await onSettled(settled); + } + 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 index 1fc52d02b56..44957fefc1b 100644 --- a/server/src/services/tool-review-work-notifications.ts +++ b/server/src/services/tool-review-work-notifications.ts @@ -4,6 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export async function notifyToolReviewChange(transaction: object, companyId: string) { await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry, companyId); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery, companyId); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewInterruption, companyId); } export function writeToolReviewChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise {