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 85c7f468f0a..286442d4b15 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2250,3 +2250,49 @@ filesystem execution proof, and retains the fake-clock connection failure and Runner CI receipt timeout. Existing lease acceptance, updated full acceptance, remote gates, older parent findings, and every remaining original duty are still required. No stack PR has been merged. + + +## Slice 31 accepted checkpoint and slice 32 owner audit + +[#17649](https://github.com/paperclipai/paperclip/pull/17649) at +`2384384eeec4f32d073242ebf597b81df2df1ccb` has all 53 checks passing, +fresh Greptile 5/5, and no inline review findings. Full chat/startup acceptance +passes all 1,505 cases on fresh `paperclip_sleep31_native_full_20261025_test` +(`/tmp/sleep31-native-full-chat.log`, 446.10 seconds, exit 0). The lease service +passes 51 tests; focused native/notice acceptance passes 36; final server +typecheck and build pass. This supersedes the earlier slice-31 pending statuses. +The initial Runner receipt timeout did not reproduce after building debug Rust +binaries (10 local tests pass), nor in a controlled 500ms event-loop-delay +experiment. The updated-head remote Runner gate passes without an unchanged-job +retry. Its exact initial cause remains unproven. + +The next branch is `codex/sleep-32-terminal-confirmations`, based on the accepted +slice-31 commit. See the terminal confirmation owner audit. There is no slice-32 +implementation or PR yet. Older #17490/#17572 CI failures, #17576 inventory +amplification findings, #17580 offboarding/cleanup ownership findings, the entire +remaining original list, final master reconciliation and all final-head checks +remain required. All stack PRs remain unmerged. + + +## Slice 32 implementation checkpoint + +`codex/sleep-32-terminal-confirmations` implements terminal-confirmation recovery +on committed signals and the shared scoped coordinator. Its Telegram owner keeps +payload validation, settlement, continuation staging and activity attribution. +The audit is `2026-10-25-terminal-confirmation-sleep-audit.md`. Nine distinct focused +cases pass across the recorded commands, including real callback failure, outer +commit/rollback, historical waiting/invalid prefixes, and a write during an active +settlement. The new source guard excludes this duty from periodic maintenance. +Full local acceptance, remote review/checks and publication are pending here. +This supersedes the audit-only slice-32 status above. All older stack gates and +every remaining original obligation remain required; no PR is merged. + +Slice 32 is now draft [#17651](https://github.com/paperclipai/paperclip/pull/17651). +Initial workspace types/build pass; full chat/startup acceptance passes 1,511 of +1,512, with a prior retry-notice test fake-clock CONNECT_TIMEOUT retained and +corrected. Initial CI passes 51 gates but fails a plugin test at a cron boundary +and its aggregate. A controlled timestamp diagnostic establishes an additional +owed cron occurrence; fixing that fixture clock passes all 28 plugin tests. +Thirteen selected chat lifecycle cases pass after the retry-notice wait fix. +Initial Greptile is 5/5 without findings. Fresh full acceptance and corrected-head +remote gates remain required; see the slice-32 audit for all retained failures. diff --git a/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md b/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md new file mode 100644 index 00000000000..1404918d846 --- /dev/null +++ b/doc/plans/2026-10-25-terminal-confirmation-sleep-audit.md @@ -0,0 +1,407 @@ +# Terminal confirmation recovery migration + +Date: 2026-10-25. Branch: `codex/sleep-32-terminal-confirmations`. +Parent: [#17649](https://github.com/paperclipai/paperclip/pull/17649) at +`2384384eeec4f32d073242ebf597b81df2df1ccb`. +Status: implemented; focused lifecycle acceptance passes. Full acceptance and remote gates are pending. + +## Obligation and existing boundary + +The owning implementation is +`server/src/services/chat-channels/interactions/confirmation-actions.ts`. +A supported Slack or Telegram confirmation callback accepts or rejects an interaction through +`issueThreadInteractionService`. Its `afterResolveInTransaction` callback +changes the exact `confirmation_response` action from `issued` to `processing` +in the same transaction as the terminal interaction decision. That action is +already the durable record of unfinished settlement. No additional table is +needed. + +`settleTerminalConfirmationAction` completes that saved decision: it marks the +action processed, retires sibling tokens, stages an actionless resolution +publication and (when required) an `interaction_wakeup`, and emits the existing +provider-attributed activity entry. It does not repeat the user's decision or +send provider traffic inside the transaction. Compare-and-set losers accept an +already processed winner, preserving callback/recovery concurrency. + +`reconcileTerminalConfirmationActions(limit)` currently scans processing actions +by updatedAt/id. It is invoked by `processPublicationMaintenance` on the periodic +chat path and by the explicit/manual publication queue path. The former is this +slice's target. Preserve the latter's manual recovery contract. Interaction +continuation delivery and GitHub maintenance remain separate reasons/PRs. + +## Producer inventory + +- [x] Acceptance callback: notify before its processing-action write inside the + existing `afterResolveInTransaction` owner. +- [x] Rejection callback: same contract, separately exercise the real path. +- [x] Initial action token creation in `chat-interaction-publications.ts` writes + `issued`; a user still deciding does not owe settlement work. It must be quiet. +- [x] Terminal/supersession publication helpers expire issued sibling tokens; + inspect their exact predicates before treating them as processing producers. +- [x] Settlement writers move processing actions to processed/failed/expired. + Preserve publication notifications, activity provenance and continuation order. +- [x] Board-resolution compatibility: current integration coverage fabricates a + processing action before the board decision. Establish how historical pending + processing rows reactivate from a later interaction change, without repeatedly + scanning an unchanged human-wait state. +- [x] Startup reconstructs processing debt. Outer rollback creates no actionable + debt, ambiguous commit uses the existing transaction signal protocol, and + first-write drain/resume retains the wake. + +## Mechanism and unresolved design checks + +Reuse the existing delivery coordinator and action worker registration rather +than adding a new scheduler or teaching app startup Telegram payload semantics. +The provider module owns settlement, malformed token handling, binding checks and +activity attribution. A shared registration may select a normalized action kind +and delegate to it. + +Do not merely move the unbounded retry behavior into a new timer. The current +settler returns false for a processing action with no conversation, and for a +still-pending interaction. The oldest unfinished prefix can therefore starve +later candidates under the fixed limit. Separate a genuinely runnable terminal +settlement from a pending interaction; audit the producer that makes the latter +runnable. Preserve explicit failure evidence for malformed bindings rather than +silently deleting work. Bound candidate work and prove progress beyond a denied +or failed prefix. Retrying a failed query is different from polling an unchanged +human decision. + +The first notification must not race callback settlement into a duplicate +publication or activity entry. Existing compare-and-set and idempotency keys are +the correctness owners. Future checks should observe the durable result and +provider-visible behavior, not only calls to a notification helper. + +## Verification plan + +The existing integration test beginning `renders safe Telegram confirmations as +opaque one-use buttons and wakes the resolved task` injects a failure after the +interaction transaction commits but before action settlement. It verifies the +saved decision, recovery publication, provider edit before continuation wake, +one-use token behavior, redelivery deduplication, board-race provenance and +failed continuation handling. Reuse its fixture and assertions rather than +building a parallel authorization fixture. + +- [x] Reproduce missed recovery after the real postcommit callback failure when + periodic/manual maintenance is not invoked. +- [x] Prove committed acceptance and rejection activate recovery from idle. +- [x] Prove first-write drain/resume, stop/restart, failed selection/retry and + writes during a drain/sweep. +- [x] Verify outer rollback/commit through actual producer ownership where the + callback service accepts a caller-owned transaction. +- [x] Prove empty state, issued tokens awaiting a person, and completed history + make no recurring SQL after settlement. +- [x] Prove bounded progress behind pending/malformed/contended candidate prefixes. +- [x] Retain the existing terminal edit-before-continuation behavior, duplicate + callback handling and board-resolution attribution. +- [x] Register production work and remove only its periodic maintenance call. +- [ ] Complete final server/workspace types/build, full chat/startup acceptance, + and fresh remote review/checks. Production wiring has a source guard; the + existing generic action-worker registry is unchanged. + +## Scope remains open + +No Cloud wake gate, transport ownership, provider deadline, or sleep eligibility +changes are implied. The supported process-local writer model remains explicit. +Every remaining original recurring-server reason and older stack finding stays +in the ledger; this audit does not reduce the requested scope. + +## Implementation and boundary decisions + +The Telegram owner now exposes only `registerTerminalConfirmationWork` through +its existing callback/service wiring. Shared app startup selects this owner; it +has no Telegram payload or decision policy. The owner reconstructs processing +action IDs from existing rows, using 100-row UUID-keyset pages, and registers +per-action entries with `registerScopedDeadlineWork`. The shared worker attempts +at most 25 entries per sweep, rotates companies, retains commits arriving during +an attempt, and backs off query/settlement failures from 30 seconds to five +minutes. A failed entry does not prevent later entries from progressing. + +Successful settlement, already terminal/missing actions, invalid actions that +receive a failure record, and actions still waiting on a pending interaction +remove their scheduled entry. The latter retain their durable processing row; +they wake when the interaction changes. Issued tokens and completed history do +not enter recovery. Missing conversation IDs now receive the existing +`confirmation_action_binding_invalid` failure classification instead of remaining +processing forever. Other invalid-binding and authorization policy stays with the +existing settler. + +Both real callback decision transactions await a confirmation intent before +writing processing state. The existing issue/company milestone notification +boundary also emits an identifier-only confirmation hint. This covers board +accept/reject, expiry, withdrawal, supersession and other supported interaction +changes without adding wrappers to each writer. It also means unrelated milestone +changes can cause one scoped reconstruction; an issue hint never restarts a global +inventory. The company/issue query includes orphaned conversation bindings in the +same company so invalid work can receive an explicit failure record. Company-wide +hints and startup reconstruct the full matching processing inventory in bounded +SQL pages; total enumeration and memory are proportional to that inventory, not +capped at 100 rows. No claim of constant startup cost is made. + +Only terminal-confirmation reconciliation was removed from periodic publication +maintenance. Explicit `processPendingPublications` maintenance still calls the +existing manual reconciler. Interaction continuation delivery and GitHub duties +are unchanged and remain separate stack reasons. The host sleep gate is unchanged, +including its treatment of durable processing debt. Process-local signals cover +supported writes in this process; independent SQL writers/other replicas require +an ingress protocol. Startup reconstruction remains the recovery path after a +process loss. + +## Evidence and retained failures + +- `origin/master` refreshed to `1949e726f8515127a04314e3793c427ccfc33c1e`; + full-stack reconciliation is still outstanding. The exact-obligation GitHub + search found no competing terminal-confirmation migration. +- `/tmp/sleep32-confirmation-before.log` is an invalid test invocation: the root + Vitest project filter `server` did not match a named project. No tests ran. +- `/tmp/sleep32-confirmation-before-valid.log` uses the server configuration and + reproduces the real callback gap: a committed action remains processing instead + of reaching processed when the new worker has no producer notification. +- `/tmp/sleep32-confirmation-lifecycle.log` initially passed accept, reject and + restart, but drain and failed-selection tests had not advanced to the retained + 30-second deadline. A failed spy assertion also leaked an unconsumed injected + error into the following legacy test. The tests now advance exactly to the + registered deadline and always restore the spy. The worker was not changed to + conceal those harness failures. +- `/tmp/sleep32-confirmation-lifecycle-fixed.log`: all six cases pass, including + the existing provider edit-before-continuation and duplicate callback journey. +- `/tmp/sleep32-confirmation-atomic.log`: eight focused cases pass, including the + startup guard and a real outer commit/rollback of the board-resolution owner. + The test passes 101 pending and 101 malformed actions, verifies failure evidence, + parks human-wait debt, and observes no fabricated provider-attributed activity. +- `/tmp/sleep32-confirmation-inflight.log`: a real PostgreSQL row lock holds one + settlement while a second real callback commits. Releasing it settles both + actions and returns the worker to quiet. This is a ninth distinct focused case. +- `/tmp/sleep32-types.log`: server typecheck passed before the final in-flight + concurrency test was added. A final typecheck is still required. +- Quiet assertions advance 120 seconds and count root database select/transaction + calls after startup and settlement. They do not establish provider suspension. + +## Broad verification and test-clock corrections + +At implementation head `aeaaed7b3e17e1377433165a38fef558acacb631`, workspace +`pnpm -r typecheck` and `pnpm build` passed. The fresh full chat/startup run passed +1,511 of 1,512 tests in 450.49 seconds (`/tmp/sleep32-full-chat.log`, database +`paperclip_sleep32_full_20261025_test`). Every new confirmation test passed. The +old failed-retry notice test's disabled case captured a real +`CONNECT_TIMEOUT localhost:55438` while its result wait advanced fake time by +30 seconds during DB work. The prior slice had corrected a different wait loop; +this remaining jump is now 10ms, with explicit advancement to the saved resume +deadline. Strict error and quiet assertions remain. + +Initial remote CI passed 51 checks but failed server shard 4/12 and the dependent +aggregate. The failed test was `plugin-scheduled-work.test.ts`, independent jobs +while completion persistence is unavailable: it expected two runs and observed +four. Direct job log: `/tmp/sleep32-ci-server4-direct.log`, run 40228214324, +job 121274198575. A controlled test clock at 19:59:55 reproduced an extra run +(`/tmp/sleep32-plugin-hour-boundary-before.log`). Added diagnostic timestamps +showed the same job's original owed timestamp at 19:59:54 and another legitimate +cron occurrence at 20:00:00, rather than replay of one occurrence +(`/tmp/sleep32-plugin-hour-boundary-diagnostic.log`). The CI log did not retain +those per-run scheduled timestamps; the controlled reproduction establishes the +failure mechanism, not every timing detail of that CI execution. + +The final plugin correction fixes its test clock at 12:10:00 before each case. +An intermediate experiment also reduced its wait steps to 10ms; that prevented +its existing 30-second startup/retry deadlines from being reached and cascaded +into fixture cleanup failures (27 plugin cases failed in +`/tmp/sleep32-check-fixes.log`). Those step changes were reverted. The same +combined run passed all 13 selected chat lifecycle cases, including the five +corrected retry-notice cases. The final one-line plugin clock fix passes all 28 +plugin scheduling tests on fresh `paperclip_sleep32_plugin_fixed_20261025_test` +(`/tmp/sleep32-plugin-fixed.log`). No production scheduling code was changed for +these two test failures. Earlier databases and logs are retained. + +The initial implementation received Greptile 5/5 with no inline findings. The +corrected head still requires a fresh full chat/startup run, final typecheck, +and new remote checks/review. No unchanged-job retry was used. The PR is +[#17651](https://github.com/paperclipai/paperclip/pull/17651), still draft. + + +## Parked confirmations: complete the interaction producer contract + +The worker intentionally removes a processing action from its deadline set while +its interaction is pending. A board decision must therefore wake that action. +Four real PostgreSQL regressions reproduced the missing wake for checkbox +accept/reject, suggestion rejection, and connection decline: the decision +committed but the action remained processing. Existing terminal publication +staging only expires issued tokens and cannot substitute for this signal. + +`notifyChatConfirmationResolution` in the existing confirmation notification +owner selects the four supported kinds: ordinary confirmations, checkbox +confirmations, task suggestions, and connection intents. The interaction service +registers this issue-scoped signal before writes in its existing transactions. +It does not add a new queue, timer, table, transaction wrapper per producer, or +broader milestone wake. Existing milestone signals remain intact and duplicate +hints in the same transaction coalesce. + +Audited producers now include: + +- Board accept/reject, suggestion acceptance/rejection, and connection resolution. +- Target revision expiry, superseded confirmation sweeps, opt-in human-comment + supersession (including historical comments), and explicit connection replacement. +- Deleted addressee, terminal issue, withdrawal, skip, and connection ownership change. +- Existing-access reconciliation in `satisfied-connection-intents.ts`, which + expires an obsolete request without inventing human consent. + +Ownership-change expiry now joins its row write and terminal chat staging in +one service-owned transaction, so the notification cannot race ahead of the +saved expiry. Pending payload changes and dismissing attention do not resolve +an interaction. Ordinary sibling creation already signals milestones; checkbox +creation does not automatically supersede an older checkbox and was not changed. + +Alternate-writer audit: tool action reviews, secret proposal reflection, tool +outcome reflection, automatic/obsolete native reviews, and native status effect +commits already publish their owning milestone signal. Native external-review +retirement runs inside that signalled status transaction. Conversation session +reset affects questions, not confirmation actions. OAuth's legacy confirmations +carry `connectionAuthorization` and are excluded from native callback token +creation; connection-intent retry phase changes remain pending. Result/summary +updates do not release a pending confirmation. + +Verification uses real Slack setup, immutable publication, and saved callback +rows. Eight board/admin cases park an action, prove quiet, resolve inside an +outer transaction, prove rollback preserves pending work, then commit and require +automatic processing/expiry and quiet again. Cases cover checkbox accept/reject, +withdrawal and opt-in comment supersession, suggestion accept/reject, and +connection decline/reassignment. The existing-access producer separately proves +its notification waits for the outer commit. Existing Telegram recovery, +restart/drain, malformed/pending history, and Slack publication regressions remain +covered. A first supersession fixture wrongly assumed creating a second checkbox +expired the first; it was corrected to the supported opt-in human-comment path +without changing product behavior. The external test DB fixture also preserves +the single-connection concurrency case; no OS semaphore cleanup is required. + +This establishes worker quiet and supported in-process reactivation. It does not +prove deployed host/database sleep or cross-replica notification coverage. + + +## Current prepared integration — October 28 + +The existing 14-path slice is normally merged onto the current prepared +failed-retry parent `c56cc27611c76587a278ac175470e0b4646b7b5f` at local revision +`c1e8542b588a4fa59b62f9829541d3daea4f1ddc`. All original patch edits and context +are identical after excluding only blob index hashes and hunk line offsets; +`/tmp/sleep32-current-parent-composition-proof.json` records the comparison. +The parent includes master `d30c8f42b8` and the inbound discovery/restoration +repairs. This preparation is local and still depends on both preceding PRs landing. + +**80 distinct cases pass:** all 16 selected chat confirmation lifecycle and +board/admin producer cases, plus 64 across the complete satisfied-connection +and app/startup suites. The chat selection skips 1,671 other cases. Fresh, +premigrated external PostgreSQL databases keep the lifecycle and connection +consumer fixtures separate. The original single-connection case remains intact. +Server typechecking, build and module boundaries pass. + +Commands/fixtures are `/tmp/sleep32-current-acceptance-command.json`; logs and +exit receipts are `/tmp/sleep32-current-{confirmations,consumers,types,build,boundaries}.*`. +Earlier failures/counts above retain their original revision scope. No current +remote review or CI, full local repository suite, live provider tenant or deployed +sleep claim follows from this local acceptance. Normal landed-parent integration +and fresh exact-head gates remain required before merge. + + +### Prepared on the published failed-retry parent + +The local normal merge `2ffeb04db0c4ec21865fc1f6d178e70b7b894a12` adopts published +failed-retry parent `fa95a1f2fb9d546aa62dbd3b43aa714f5c88bd35`, after the inbound +slice merged into actual master. The complete 14-path owned binary patch remains +identical across that integration; the parent carries current monitor/drain and +durable comment-write changes. Proof: +`/tmp/sleep32-published-parent-composition-proof.json`. + +**29 current cases pass:** all 16 selected confirmation producer/lifecycle cases +(1,671 skipped) and all 13 app/startup cases. Types, build and module boundaries +pass. Commands, actual fixture and revision are in +`/tmp/sleep32-published-parent-acceptance-command.json`; logs/exit receipts are +`/tmp/sleep32-published-parent-{types,build,boundaries,confirmations,startup}.*`. +The earlier 80-case matrix retains its original revision scope. This slice is +still local and unpublished: its failed-retry parent must merge before final +landed-parent integration and fresh remote checks/review. The following +acceptance-only edit leaves all owned server sources/tests unchanged. + + +### Acceptance on the actual landed retry parent + +Failed-retry notices #17649 have merged at +`f4f02880cae4f5fa54d47d39565ca99b6b72446d`. The final normal two-parent integration +is `32e64a7e9ea5b457e54ca7b268be90e1e155b451`. Its semantic merge base is verified +parent `fa95a1f2fb9d546aa62dbd3b43aa714f5c88bd35`, resolving historical conflicts +caused by squash ancestry. The complete 14-path owned binary patch is identical +and all five unrelated current-master paths are preserved. No rebase/force push +is used; the normal merge parents and exact tree are verified. + +The same **29 distinct confirmation/startup cases pass again on actual master**: +16 selected confirmation cases (1,671 skipped) and all 13 startup cases. These +are repeated cases, not 58 different tests. Server typechecking/build and module +boundaries pass before testing. Commands, fixtures and revision are in +`/tmp/sleep32-landed-acceptance-command.json`; receipts/logs are +`/tmp/sleep32-landed-{types,build,boundaries,confirmations,startup}.*`. +The earlier 80-case matrix retains its original revision scope. All 12 owned +server source/test hashes remain unchanged after the acceptance-only follow-up, +recorded in `/tmp/sleep32-landed-tested-source-sha256.json`; the integration proof +is `/tmp/sleep32-landed-parent-publication-proof.json`. Publication requires fresh +exact-head remote CI and Greptile 5/5. No physical sleep or full local repository +result is inferred from these checks. + + +### Bound retained-history reads — October 28 review repair + +Greptile found a valid P2 even while reporting 5/5: the new event-driven +reconstruction still examined retained completed actions. A returned-row limit +alone does not limit database work. This finding is fixed before merge. + +The generated `0398_yielding_random.sql` migration adds a partial index on +`chat_actions(company_id, id)` only for processing `confirmation_response` rows. +It uses `CREATE INDEX CONCURRENTLY IF NOT EXISTS`, registered with the existing +top-level migration runner. Existing SQL/journal history and retained snapshot +bytes are preserved; generation prunes snapshot 0393 and retains 0394–0398. +This adds no table or scheduler service. + +The index alone was insufficient. PostgreSQL still preferred the primary key +for global `ORDER BY id`, examining 20,206 action rows and hitting 20,276 buffers +in the recorded plan. Reconstruction now orders and keyset-pages by +`(company_id, id)`, matching the one sparse index for startup and scoped reads. +The fixed kind/status predicates remain SQL literals so their partial-index +eligibility does not depend on parameter values in a prepared plan. Selection +continues to page 100 rows and keeps the existing company-fair dispatch owner. + +The regression captures actual emitted discovery SQL and executes it unchanged +under `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)`. With 20,000 completed/unrelated +actions and one pending decision, the original query failed at 20,009 examined +rows. Both final startup and company-scoped queries must return the correct +pending action, examine at most two action rows, and touch fewer than 20 action +buffers. The empty scoped query must also stay below 20 buffers. Planner +settings are unchanged. The fixture vacuums/analyzes its owned table first to +separate retained live history from dead index entries left by prior tests. +The existing 101 pending/101 malformed prefix case also passes with composite +paging and a terminal action in another company. + +**110 distinct cases pass:** 17 final confirmation lifecycle/query-cost cases +(1,671 skipped), seven real Runner environment-prepared descriptor cases +(295 skipped), and all 86 cases across app/startup, bundled plugins and plugin +database suites. The last two groups run on the same product source tree; +subsequent fixture hardening changes only the cost test's vacuum precondition, +and all 17 confirmation cases and server typechecking pass again afterward. +Server/database typechecking, server/database build, migration safety/numbering, +module boundaries and a fresh PostgreSQL migration through 0398 pass. + +Normal integration `693e6b22536104fe7be5cebe959e14b295f27dc0` adopts master +`644937c93150c423f12674992c7ea1e8b5dbc98c` (boxd provider/plugin integration). +The complete owned binary patch and all 18 unrelated master paths are preserved; +`/tmp/sleep32-index-master-composition-proof.json` records that comparison. +Commands and receipts: `/tmp/sleep32-index-master-acceptance-command.json`, +`/tmp/sleep32-index-master-{db-build,types,db-types,build,boundaries,upgrade,confirmations,runner,startup}.*`, +and `/tmp/sleep32-index-final-{types,confirmations}.*`. + +Failure history remains visible. The initial generated-index attempt passed +16 lifecycle cases but its new scoped observation used a one-second wait against +the owner's existing one-second coalescing interval; that test wait is now five +seconds, without changing product cadence or the 15-second case budget. The +subsequent index-only plan still scanned completed history, motivating aligned +keyset order. The preceding published head's Runner shard failed its +`timeout-attachment` durable-suspension case (1 failed/1,955 passed/4 skipped). +All seven descriptor variants pass locally unchanged on this repair, but only +fresh exact-head remote CI can establish that shard's current pass. Old failed +logs are retained in `/tmp/sleep32-runner-ci-failed.log`; no failed gate is waived. +Fresh Greptile 5/5 with no unresolved actionable finding and all remote checks +remain required. These results do not establish physical host/database sleep. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 9f583812d56..318c9f09317 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -549,6 +549,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0395_tranquil_firestar.sql", "0396_tan_paladin.sql", "0397_known_madrox.sql", + "0398_yielding_random.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0398_yielding_random.sql b/packages/db/src/migrations/0398_yielding_random.sql new file mode 100644 index 00000000000..b2dc333339e --- /dev/null +++ b/packages/db/src/migrations/0398_yielding_random.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_confirmation_recovery_idx" ON "chat_actions" USING btree ("company_id","id") WHERE "chat_actions"."kind" = 'confirmation_response' AND "chat_actions"."status" = 'processing'; diff --git a/packages/db/src/migrations/meta/0393_snapshot.json b/packages/db/src/migrations/meta/0398_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0393_snapshot.json rename to packages/db/src/migrations/meta/0398_snapshot.json index 70b260ad9aa..2d38694b8c3 100644 --- a/packages/db/src/migrations/meta/0393_snapshot.json +++ b/packages/db/src/migrations/meta/0398_snapshot.json @@ -1,6 +1,6 @@ { - "id": "e1ddc6f7-6335-4b3d-aeec-a5df27793874", - "prevId": "6974dd25-b8e9-4143-b4b7-ee5a5c3ff54a", + "id": "7f056258-4b1a-4828-b950-33f070fc0c47", + "prevId": "c7947622-f079-44c9-a57a-5a11d1bec5e8", "version": "7", "dialect": "postgresql", "tables": { @@ -1350,6 +1350,34 @@ "concurrently": false, "method": "btree", "with": {} + }, + "activity_log_company_user_issue_actor_idx": { + "name": "activity_log_company_user_issue_actor_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "actor_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "entity_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"activity_log\".\"actor_type\" = 'user' AND \"activity_log\".\"entity_type\" = 'issue'", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -9596,6 +9624,28 @@ } }, "indexes": { + "chat_actions_confirmation_recovery_idx": { + "name": "chat_actions_confirmation_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'confirmation_response' AND \"chat_actions\".\"status\" = 'processing'", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_actions_provider_action_uq": { "name": "chat_actions_provider_action_uq", "columns": [ @@ -14160,6 +14210,24 @@ "primaryKey": false, "notNull": false }, + "provider_account_id": { + "name": "provider_account_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "cleanup_confirmed_at": { + "name": "cleanup_confirmed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "cleanup_next_attempt_at": { + "name": "cleanup_next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "expires_at": { "name": "expires_at", "type": "timestamp with time zone", @@ -14182,6 +14250,22 @@ } }, "indexes": { + "chat_voice_inbound_cleanup_idx": { + "name": "chat_voice_inbound_cleanup_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_voice_inbound_expiry_idx": { "name": "chat_voice_inbound_expiry_idx", "columns": [ @@ -14934,6 +15018,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_voice_sessions_report_window_idx": { + "name": "chat_voice_sessions_report_window_idx", + "columns": [ + { + "expression": "ended_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -42987,6 +43093,99 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.native_runner_checkpoints": { + "name": "native_runner_checkpoints", + "schema": "", + "columns": { + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "issue_id": { + "name": "issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "part": { + "name": "part", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "material": { + "name": "material", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "native_runner_checkpoints_headers_idx": { + "name": "native_runner_checkpoints_headers_idx", + "columns": [ + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_runner_checkpoints\".\"part\" = 'header'", + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk": { + "name": "native_runner_checkpoints_company_id_issue_id_run_id_heartbeat_runs_company_id_native_issue_id_id_fk", + "tableFrom": "native_runner_checkpoints", + "tableTo": "heartbeat_runs", + "columnsFrom": [ + "company_id", + "issue_id", + "run_id" + ], + "columnsTo": [ + "company_id", + "native_issue_id", + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "native_runner_checkpoints_run_id_part_pk": { + "name": "native_runner_checkpoints_run_id_part_pk", + "columns": [ + "run_id", + "part" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.pipeline_automation_executions": { "name": "pipeline_automation_executions", "schema": "", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index b402ee53314..b9355079da4 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2773,6 +2773,13 @@ "when": 1793195976732, "tag": "0397_known_madrox", "breakpoints": true + }, + { + "idx": 398, + "version": "7", + "when": 1793216165150, + "tag": "0398_yielding_random", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_channels.ts b/packages/db/src/schema/chat_channels.ts index 9f83918b3d8..b6aeb4c9efd 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -673,6 +673,9 @@ export const chatActions = pgTable( .defaultNow(), }, (table) => [ + index("chat_actions_confirmation_recovery_idx") + .on(table.companyId, table.id) + .where(sql`${table.kind} = 'confirmation_response' AND ${table.status} = 'processing'`), uniqueIndex("chat_actions_provider_action_uq").on( table.endpointId, table.providerActionId, diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 1dad4ca398a..629bd1139f6 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -92,6 +92,8 @@ describe("createChatReconciliationCoordinator", () => { chat.indexOf("async function schedulePendingPublications(")); expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications("); expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications("); + expect(publicationMaintenance).not.toContain("reconcileTerminalConfirmationActions("); + expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,"); expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,"); }); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 59856338ce7..c0f6c6494fb 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -41873,6 +41873,99 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await service.shutdown(); }); + it.each([ + ["request_checkbox_confirmation", "accept"], + ["request_checkbox_confirmation", "reject"], + ["request_checkbox_confirmation", "withdraw"], + ["request_checkbox_confirmation", "supersede"], + ["suggest_tasks", "accept"], + ["suggest_tasks", "reject"], + ["connection_intent", "reject"], + ["connection_intent", "reassign"], + ] as const)("wakes parked Slack %s actions after board %s commits", async (kind, decision) => { + const previousPublicUrl = process.env.PAPERCLIP_PUBLIC_URL; + process.env.PAPERCLIP_PUBLIC_URL = "https://paperclip.example"; + const fixture = await seedCompany(); + const { callbacks, endpoint, service } = await configuredSlackEndpoint(fixture); + let coordinator: ReturnType | undefined; + try { + const thread = makeThread({ channelId: "C-REVIEW", id: "slack:C-REVIEW:4500.1", name: "reviews" }); + await deliverMessage({ callbacks, endpointId: endpoint.id, thread: thread.thread, message: makeMessage({ id: "4500.1", text: "@maya review this change", mentioned: true, userId: `U-REVIEW-${randomUUID()}` }), trigger: "mention" }); + await qualifySetupRoundTrip(service, endpoint.id); + await service.test(endpoint.id, "owner-user"); + const [conversation] = await readChatConversationsByEndpointId(endpoint.id); + const sourceRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId: fixture.companyId, agentId: fixture.assignedAgentId, status: "succeeded", responsibleUserId: "owner-user", contextSnapshot: { issueId: conversation.issueId, taskId: conversation.issueId, source: "automation" } }); + const interactions = issueThreadInteractionService(db); + const issue = { id: conversation.issueId, companyId: fixture.companyId }; + const interaction = kind === "connection_intent" + ? await interactions.createConnectionIntent(issue, { sourceRunId, addresseeUserId: "owner-user", idempotencyKey: `slack-review:${sourceRunId}`, payload: { version: 1, serviceSlug: "notion", serviceName: "Notion", requestingAgentId: fixture.assignedAgentId, requestingAgentName: "Maya", phase: "requested" } }) + : await interactions.create(issue, { kind, sourceRunId, payload: kind === "suggest_tasks" ? { version: 1, tasks: [{ clientKey: "docs", title: "Review migration notes" }] } : { version: 1, prompt: "Review migration notes", ...(kind === "request_checkbox_confirmation" ? { options: [{ id: "docs", label: "Migration notes" }], supersedeOnUserComment: decision === "supersede" } : {}) } } as Parameters[1], { agentId: fixture.assignedAgentId }); + await service.processPendingPublications(); + const rows = () => db.select().from(chatPublications).where(and(eq(chatPublications.endpointId, endpoint.id), eq(sql`${chatPublications.payload}->>'interactionId'`, interaction.id))); + const [original] = await rows(); + expect(original?.state).toBe("published"); + expect(original?.payload.card?.actions).toEqual(expect.arrayContaining([expect.objectContaining({ type: "callback" })])); + + const [action] = await db.select().from(chatActions).where(and( + eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "confirmation_response"), + eq(sql`${chatActions.payload}->>'interactionId'`, interaction.id), + eq(sql`${chatActions.payload}->>'decision'`, decision === "accept" ? "accept" : "reject"), + )); + expect(action).toBeDefined(); + await db.update(chatActions).set({ status: "processing" }).where(eq(chatActions.id, action.id)); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const quiet = async () => { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + }; + await service.registerTerminalConfirmationWork(coordinator, () => true).ready; + await settle(); await quiet(); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].status).toBe("processing"); + const resolve = async (tx: Db) => { + const board = issueThreadInteractionService(tx); + if (decision === "withdraw") await board.withdrawInteraction(issue, interaction.id, {}, { userId: "owner-user" }); + else if (decision === "reassign") await board.expireConnectionIntentsForOwnershipChange(issue); + else if (decision === "supersede") { + const [comment] = await tx.insert(issueComments).values({ companyId: issue.companyId, issueId: issue.id, + authorUserId: "owner-user", body: "Use the revised migration plan", createdAt: new Date(Date.now() + 1000) }).returning(); + await board.expireRequestConfirmationsSupersededByComment(issue, comment, { userId: "owner-user" }); + } + else if (kind === "connection_intent") await board.resolveConnectionIntent(issue, interaction.id, + { version: 1, outcome: "declined" }, { userId: "owner-user" }); + else if (decision === "accept") await board.acceptInteraction(issue, interaction.id, + { selectedOptionIds: ["docs"], selectedClientKeys: ["docs"] }, { userId: "owner-user" }); + else await board.rejectInteraction(issue, interaction.id, {}, { userId: "owner-user" }); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].status).toBe("processing"); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0].status).toBe("pending"); + }; + await expect(db.transaction(async tx => { await resolve(tx as unknown as Db); throw new Error("rollback board decision"); })) + .rejects.toThrow("rollback board decision"); + await settle(); await quiet(); + await db.transaction(tx => resolve(tx as unknown as Db)); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].status).toBe(["accept", "reject"].includes(decision) ? "processed" : "expired"); + }, { timeout: 5_000 }); + await settle(); await quiet(); + expect(errors).toEqual([]); + } finally { + await coordinator?.stop(); vi.useRealTimers(); await service.shutdown(); + if (previousPublicUrl === undefined) delete process.env.PAPERCLIP_PUBLIC_URL; + else process.env.PAPERCLIP_PUBLIC_URL = previousPublicUrl; + } + }); + it.each(["request_confirmation", "request_checkbox_confirmation", "request_item_verdicts", "suggest_tasks", "connection_intent"] as const)("delivers and replaces a Slack %s card through the publication service", async kind => { const previousPublicUrl = process.env.PAPERCLIP_PUBLIC_URL; process.env.PAPERCLIP_PUBLIC_URL = "https://paperclip.example"; @@ -45407,7 +45500,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(stillPending.status).toBe("pending"); }); - it("renders safe Telegram confirmations as opaque one-use buttons and wakes the resolved task", async () => { + async function telegramConfirmationRecoveryFixture(decision: "accept" | "reject" = "accept") { const fixture = await seedCompany(); let injectSettlementCrash = true; const settlementBarrier = vi.fn(async () => { @@ -45565,7 +45658,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ).toBe(true); const acceptAction = confirmationActions.find( - (action) => action.payload.decision === "accept", + (action) => action.payload.decision === decision, ); if (!acceptAction) throw new Error("Telegram accept action was not persisted"); @@ -45597,7 +45690,286 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { value: undefined, }, }; - await expect(callbacks.onAction(actionEvent)).rejects.toThrow( + return { fixture, callbacks, onAction: callbacks.onAction, endpoint, service, wakeup, settlementBarrier, dm, conversation, linkedUserId, principal, interaction, publication, acceptAction, actionEvent }; + } + + it.each(["accept", "reject", "drain", "restart", "failed selection"] as const)("recovers committed terminal confirmations without polling through %s", async mode => { + const context = await telegramConfirmationRecoveryFixture(mode === "reject" ? "reject" : "accept"); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const quiet = async () => { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + }; + try { + await context.service.registerTerminalConfirmationWork(coordinator, () => canRun).ready; + await settle(); await quiet(); + if (["drain", "failed selection"].includes(mode)) canRun = false; + if (mode === "restart") await coordinator.stop(); + await expect(context.onAction(context.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + if (["drain", "failed selection", "restart"].includes(mode)) { + await vi.advanceTimersByTimeAsync(1000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, context.acceptAction.id))).toEqual([{ status: "processing" }]); + await quiet(); + if (mode === "failed selection") { + const select = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("temporary confirmation selection failure"); }); + try { + canRun = true; + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(errors).toHaveLength(1); }); + canRun = false; + } finally { select.mockRestore(); } + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + await quiet(); + } + canRun = true; + if (coordinator.nextWakeAt() !== null) await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + await context.service.registerTerminalConfirmationWork(coordinator, () => canRun).ready; + } + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, context.acceptAction.id))).toEqual([{ status: "processed" }]); + }, { timeout: 10_000 }); + await settle(); await quiet(); + expect(await db.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.companyId, context.fixture.companyId), eq(chatActions.kind, "interaction_wakeup"), + eq(sql`${chatActions.payload}->>'interactionId'`, context.interaction.id), + ))).toHaveLength(1); + expect(await db.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, context.fixture.companyId), + eq(sql`${activityLog.details}->>'interactionId'`, context.interaction.id), + eq(sql`${activityLog.details}->>'source'`, "external_chat"), + ))).toHaveLength(1); + expect(errors).toEqual(mode === "failed selection" ? [expect.objectContaining({ + errors: [expect.objectContaining({ message: "temporary confirmation selection failure" })], + })] : []); + } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("parks pending confirmation history, passes malformed prefixes, and wakes on the outer interaction commit", async () => { + const waiting = await telegramConfirmationRecoveryFixture(); + const terminal = await telegramConfirmationRecoveryFixture(); + const pendingIds = Array.from({ length: 101 }, (_, index) => `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`); + const invalidIds = Array.from({ length: 101 }, (_, index) => `00000000-0000-4000-8001-${String(index + 1).padStart(12, "0")}`); + await db.insert(chatActions).values([...pendingIds, ...invalidIds].map(id => ({ + ...waiting.acceptAction, id, providerActionId: `confirmation-history:${id}`, status: "processing", + conversationId: invalidIds.includes(id) ? null : waiting.conversation.id, result: null, + }))); + await expect(terminal.onAction(terminal.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 30_000 }); + const quiet = async () => { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + }; + try { + await waiting.service.registerTerminalConfirmationWork(coordinator, () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, terminal.acceptAction.id))).toEqual([{ status: "processed" }]); + }, { timeout: 10_000 }); + await settle(); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(101); + expect(await db.select({ id: chatActions.id }).from(chatActions).where(and( + inArray(chatActions.id, invalidIds), eq(chatActions.status, "failed"), + eq(sql`${chatActions.result}->>'code'`, "confirmation_action_binding_invalid"), + ))).toHaveLength(101); + await quiet(); + const [issue] = await db.select().from(issues).where(eq(issues.id, waiting.conversation.issueId)); + await expect(db.transaction(async tx => { + await issueThreadInteractionService(tx as unknown as Db).acceptInteraction(issue, waiting.interaction.id, {}, { userId: waiting.linkedUserId }); + expect(await db.select({ status: issueThreadInteractions.status }).from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, waiting.interaction.id))).toEqual([{ status: "pending" }]); + throw new Error("rollback outer confirmation decision"); + })).rejects.toThrow("rollback outer confirmation decision"); + await settle(); await quiet(); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(101); + await db.transaction(async tx => { + await issueThreadInteractionService(tx as unknown as Db).acceptInteraction(issue, waiting.interaction.id, {}, { userId: waiting.linkedUserId }); + expect(await db.select({ status: issueThreadInteractions.status }).from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, waiting.interaction.id))).toEqual([{ status: "pending" }]); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(101); + }); + await settle(); await quiet(); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processing")))).toHaveLength(0); + expect(await db.select({ id: chatActions.id }).from(chatActions) + .where(and(inArray(chatActions.id, pendingIds), eq(chatActions.status, "processed")))).toHaveLength(1); + expect(await db.select({ id: activityLog.id }).from(activityLog).where(and( + eq(activityLog.companyId, waiting.fixture.companyId), + eq(sql`${activityLog.details}->>'interactionId'`, waiting.interaction.id), + eq(sql`${activityLog.details}->>'source'`, "external_chat"), + ))).toHaveLength(0); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await db.delete(chatActions).where(inArray(chatActions.id, [...pendingIds, ...invalidIds])); + } + }); + + it("bounds terminal confirmation recovery reads with completed history", async () => { + const waiting = await telegramConfirmationRecoveryFixture(); + const prefix = `confirmation-plan:${randomUUID()}:`; + await db.execute(sql` + INSERT INTO chat_actions (company_id, endpoint_id, kind, provider_action_id, status) + SELECT ${waiting.fixture.companyId}::uuid, ${waiting.endpoint.id}::uuid, + CASE WHEN n <= 10000 THEN 'confirmation_response' ELSE 'unrelated_action' END, + ${prefix} || n, + CASE WHEN n <= 10000 THEN 'processed' ELSE 'processing' END + FROM generate_series(1, 20000) AS n + `); + await db.update(chatActions).set({ status: "processing", result: null }) + .where(eq(chatActions.id, waiting.acceptAction.id)); + await db.execute(sql`VACUUM (ANALYZE) chat_actions`); + const statements: { sql: string; params: unknown[] }[] = []; + const select = db.select.bind(db); + const reads = vi.spyOn(db, "select").mockImplementation((...args) => { + const selection = select(...args), from = selection.from.bind(selection); + vi.spyOn(selection, "from").mockImplementation(table => { + const query = from(table); + if (table === chatActions) { + const execute = query.execute.bind(query); + vi.spyOn(query, "execute").mockImplementation(async () => { + const statement = query.toSQL(); + if (statement.sql.includes("'confirmation_response'") && statement.sql.includes("'processing'")) { + statements.push(statement); + } + return execute(); + }); + } + return query; + }); + return selection; + }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + try { + await waiting.service.registerTerminalConfirmationWork(coordinator, () => true).ready; + await vi.waitFor(() => expect(statements.length).toBeGreaterThan(0)); + const startup = statements[0]; + const beforeHint = statements.length; + await db.transaction(async tx => { + await signalDatabaseWork(tx, DELIVERY_QUEUES.chatTerminalConfirmations, + JSON.stringify({ companyId: waiting.fixture.companyId })); + }); + await vi.waitFor(() => expect(statements.length).toBeGreaterThan(beforeHint), { timeout: 5_000 }); + const scoped = statements.at(-1)!; + expect(scoped.params).toContain(waiting.fixture.companyId); + await coordinator.stop(); + reads.mockRestore(); + for (const statement of [startup, scoped]) { + expect(await db.$client.unsafe(statement.sql, statement.params as Parameters[1])) + .toMatchObject([{ id: waiting.acceptAction.id }]); + const result = await db.$client.unsafe( + `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${statement.sql}`, statement.params as Parameters[1], + ); + const plan = result[0]["QUERY PLAN"][0].Plan; + type PlanNode = { Plans?: PlanNode[]; [key: string]: unknown }; + const nodes = (node: PlanNode): PlanNode[] => [node, ...(node.Plans ?? []).flatMap(nodes)]; + const actionReads = nodes(plan).filter(node => node["Relation Name"] === "chat_actions"); + expect(actionReads).toHaveLength(1); + expect(Number(actionReads[0]["Actual Rows"]) + Number(actionReads[0]["Rows Removed by Filter"] ?? 0), + JSON.stringify({ params: statement.params, actionReads })) + .toBeLessThanOrEqual(2); + expect(Number(actionReads[0]["Shared Hit Blocks"]) + Number(actionReads[0]["Shared Read Blocks"])) + .toBeLessThan(20); + } + await db.update(chatActions).set({ status: "processed" }).where(eq(chatActions.id, waiting.acceptAction.id)); + await db.execute(sql`ANALYZE chat_actions`); + expect(await db.$client.unsafe(scoped.sql, scoped.params as Parameters[1])).toHaveLength(0); + const empty = await db.$client.unsafe( + `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${scoped.sql}`, scoped.params as Parameters[1], + ); + expect(empty[0]["QUERY PLAN"][0].Plan["Shared Hit Blocks"] + + empty[0]["QUERY PLAN"][0].Plan["Shared Read Blocks"]).toBeLessThan(20); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); reads.mockRestore(); + await db.delete(chatActions).where(sql`${chatActions.providerActionId} LIKE ${`${prefix}%`}`); + } + }); + + it("retains a confirmation committed while another settlement is in flight", async () => { + const first = await telegramConfirmationRecoveryFixture(); + const second = await telegramConfirmationRecoveryFixture(); + await expect(first.onAction(first.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + let locked!: () => void, release!: () => void; + const lockReady = new Promise(resolve => { locked = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + const transaction = db.transaction(async tx => { + await tx.select({ id: chatActions.id }).from(chatActions).where(eq(chatActions.id, first.acceptAction.id)).for("update"); + locked(); await held; + }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let ready: Promise | undefined; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await lockReady; + ready = first.service.registerTerminalConfirmationWork(coordinator, () => true).ready; + await vi.waitFor(async () => { + const blocked = await db.execute(sql`select pid from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and query like 'update "chat_actions"%'`); + expect(blocked.length).toBeGreaterThan(0); + }, { timeout: 10_000 }); + await expect(second.onAction(second.actionEvent)).rejects.toThrow("injected post-interaction confirmation settlement crash"); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, second.acceptAction.id))).toEqual([{ status: "processing" }]); + release(); await transaction; await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(await db.select({ status: chatActions.status }).from(chatActions) + .where(inArray(chatActions.id, [first.acceptAction.id, second.acceptAction.id]))) + .toEqual([{ status: "processed" }, { status: "processed" }]); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const select = vi.spyOn(db, "select"), queryTransaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(queryTransaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); queryTransaction.mockRestore(); } + expect(errors).toEqual([]); + } finally { + release(); await transaction; await ready; await coordinator.stop(); vi.useRealTimers(); + } + }); + + it("renders safe Telegram confirmations as opaque one-use buttons and wakes the resolved task", async () => { + const { fixture, onAction, endpoint, service, wakeup, settlementBarrier, dm, conversation, linkedUserId, principal, interaction, publication, acceptAction, actionEvent } = await telegramConfirmationRecoveryFixture(); + await expect(onAction(actionEvent)).rejects.toThrow( "injected post-interaction confirmation settlement crash", ); expect(settlementBarrier).toHaveBeenCalledTimes(1); @@ -45776,7 +46148,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ) .then((rows) => rows.length); - await callbacks.onAction(actionEvent); + await onAction(actionEvent); await new Promise((resolve) => setImmediate(resolve)); expect(dm.post).toHaveBeenCalledTimes(providerPostCountBeforeRedelivery); await expect( @@ -64915,6 +65287,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); await quiet(); canRun = true; + if (coordinator.nextWakeAt() !== null) await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); if (mode === "stopped") { await context.service.processPublicationMaintenance(); expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); @@ -64924,7 +65297,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } } await vi.waitFor(async () => { - await vi.advanceTimersByTimeAsync(30_000); + await vi.advanceTimersByTimeAsync(10); expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ payload: mode === "disabled" ? { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" } diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index 36eb4555df3..54e4d934aae 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { enqueueIssueInteractionChatPublications } from "../services/chat-interaction-publications.js"; import { randomUUID } from "node:crypto"; import { eq, sql } from "drizzle-orm"; @@ -20,16 +21,20 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; import { reserveRunBudget } from "../services/budget-reservations.js"; -const support = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SATISFIED_CONNECTION_TEST_DATABASE_URL"); +const support = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; describePostgres("already available connection requests", () => { let db!: ReturnType; let temp!: Awaited>; beforeAll(async () => { - temp = await startEmbeddedPostgresTestDatabase("paperclip-satisfied-connections-"); - db = createDb(temp.connectionString); - await db.insert(authUsers).values({ id: "connection-owner", name: "Owner", email: "owner@example.test", createdAt: new Date(), updatedAt: new Date() }); + if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); + else { + temp = await startEmbeddedPostgresTestDatabase("paperclip-satisfied-connections-"); + db = createDb(temp.connectionString); + } + await db.insert(authUsers).values({ id: "connection-owner", name: "Owner", email: "owner@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); }, 20_000); beforeEach(async () => { await db.execute(sql`truncate table companies cascade`); }); afterAll(async () => { await db.$client.end({ timeout: 1 }); await temp?.cleanup(); }); @@ -60,6 +65,21 @@ describePostgres("already available connection requests", () => { return run!; } + it("notifies terminal confirmation recovery after existing-access retirement commits", async () => { + const f = await seed(); + const notified = vi.fn(); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatTerminalConfirmations, notified); + try { + await db.transaction(async tx => { + expect(await satisfiedConnectionIntentService(tx as unknown as typeof db).sweepPending()).toMatchObject({ satisfied: 1 }); + expect(notified).not.toHaveBeenCalled(); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, f.intent.id)))[0].status).toBe("pending"); + }); + expect(notified).toHaveBeenCalledOnce(); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, f.intent.id)))[0].status).toBe("expired"); + } finally { unsubscribe(); } + }); + it("settles a published Slack card when existing access retires the connection request", async () => { const f = await seed(); const endpointId = randomUUID(); @@ -378,7 +398,7 @@ describePostgres("already available connection requests", () => { it("records a system completion and one durable continuation with a one-connection pool", async () => { const f = await seed(); - const single = createDb(temp.connectionString, { maxConnections: 1 }); + const single = createDb(externalTestDatabaseUrl ?? temp.connectionString, { maxConnections: 1 }); try { const outcomes = await Promise.all([satisfiedConnectionIntentService(single).sweepPending(), satisfiedConnectionIntentService(db).sweepPending()]); expect(outcomes.reduce((sum, outcome) => sum + outcome.satisfied, 0)).toBe(1); diff --git a/server/src/app.ts b/server/src/app.ts index 09ac7213ef6..2805f00b2a0 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1288,6 +1288,7 @@ export async function createApp( chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 1675fd413e0..4e504e2b8f7 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -3350,7 +3350,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPublicationMaintenance(limit = 25) { scheduleGitHubMaintenance(); - await reconcileTerminalConfirmationActions(limit); await processPendingInteractionWakeups(limit); } @@ -3479,7 +3478,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { processPendingInteractionWakeups } = createChatInteractionContinuations(db, { redactError, CONFIRMATION_WAKEUP_RETRY_BACKOFF_MS, get heartbeat() { return options.heartbeat; }, }); - const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions } = createChatInteractionCallbacks(db, { + const { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork } = createChatInteractionCallbacks(db, { heartbeat: options.heartbeat, CAPABILITIES, MAX_ERROR_TEXT, actionThreadMatchesConversation, nonDirectDestinationAllowed, sameTeamsThreadIdentity, ensurePrincipal, endpointRecord, runtimeCallbackRecord, runtimeCallbackEndpoint, @@ -3503,6 +3502,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerTerminalConfirmationWork, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices, registerTeamsAttemptRecoveryWork, diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 333b3df379c..24dfb13fede 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -145,7 +145,7 @@ type InteractionCallbacksOptions = { export function createChatInteractionCallbacks(db: Db, options: InteractionCallbacksOptions) { const { CAPABILITIES, MAX_ERROR_TEXT, actionThreadMatchesConversation, nonDirectDestinationAllowed, sameTeamsThreadIdentity, ensurePrincipal, endpointRecord, runtimeCallbackRecord, runtimeCallbackEndpoint, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, recordCurrentMicrosoftTeamsRoute, lockTelegramCallbackNoticeAuthority, stageProviderEffect, scheduleProviderEffect, invalidateRuntime, redactError, persistence, questionResponses, scheduleMessageProcessing, processPendingPublications } = options; const { handleDiscordFormCorrection, acceptDiscordQuestionForm } = createDiscordQuestionForms(db, { requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, persistence }); - const { reconcileTerminalConfirmationActions, handleChatConfirmation } = createChatConfirmationActions(db, { + const { reconcileTerminalConfirmationActions, handleChatConfirmation, registerTerminalConfirmationWork } = createChatConfirmationActions(db, { heartbeat: options.heartbeat, requireCurrentExternalActionAuthorization, isExternalActionAuthorizationChange, scheduleMessageProcessing, processPendingPublications, get confirmationResolutionPersistBarrier() { return options.confirmationResolutionPersistBarrier; }, }); @@ -2029,5 +2029,5 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb }); return acceptedResponse(); } - return { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions }; + return { handleAction, handleModalSubmit, reconcileTerminalConfirmationActions, registerTerminalConfirmationWork }; } diff --git a/server/src/services/chat-channels/interactions/confirmation-actions.ts b/server/src/services/chat-channels/interactions/confirmation-actions.ts index caaef9ab99f..91c60570b1d 100644 --- a/server/src/services/chat-channels/interactions/confirmation-actions.ts +++ b/server/src/services/chat-channels/interactions/confirmation-actions.ts @@ -3,12 +3,20 @@ import { queueIssueAssignmentWakeup, type IssueAssignmentWakeupDeps } from "../. import { chatInteractionPresentation } from "../../chat-interaction-presentation.js"; import type { DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import { notifyChatConfirmationWork, type ChatConfirmationScope } from "../../chat-confirmation-notifications.js"; + import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; import { and, asc, eq, + gt, + isNull, + or, inArray, ne, sql, @@ -176,6 +184,7 @@ export function createChatConfirmationActions(db: Db, options: ChatConfirmationO }); }, afterResolveInTransaction: async (tx: DbTransaction, resolved: import("@paperclipai/shared").IssueThreadInteraction) => { + await notifyChatConfirmationWork(tx, { companyId: action.companyId, issueId: conversation.issueId }); const [owned] = await tx .update(chatActions) .set({ @@ -286,8 +295,13 @@ export function createChatConfirmationActions(db: Db, options: ChatConfirmationO ) .then((rows) => rows[0] ?? null); if (action?.status === "processed") return true; - if (!action || action.status !== "processing" || !action.conversationId) + if (!action || action.status !== "processing") return false; + if (!action.conversationId) { + await db.update(chatActions).set({ + status: "failed", result: { code: "confirmation_action_binding_invalid" }, updatedAt: new Date(), + }).where(and(eq(chatActions.id, action.id), eq(chatActions.status, "processing"))); return false; + } const token = action.payload; if ( token.version !== 1 || @@ -647,5 +661,44 @@ export function createChatConfirmationActions(db: Db, options: ChatConfirmationO } return settled; } - return { settleTerminalConfirmationAction, reconcileTerminalConfirmationActions, handleChatConfirmation }; + function registerTerminalConfirmationWork( + coordinator: ReturnType, + canRun: () => boolean, + ) { + type Entry = { id: string; companyId: string; issueId: string | null; dueAt: number }; + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatTerminalConfirmations, + decodeScope: (hint: string): ChatConfirmationScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: ChatConfirmationScope) => !scope || + (entry.companyId === scope.companyId && (!scope.issueId || !entry.issueId || entry.issueId === scope.issueId)), + load: async (scope?: ChatConfirmationScope, signal?: AbortSignal): Promise => { + const entries: Entry[] = []; + let cursor: { companyId: string; id: string } | undefined; + while (!signal?.aborted) { + const rows = await db.select({ + id: chatActions.id, companyId: chatActions.companyId, issueId: chatConversations.issueId, + }).from(chatActions).leftJoin(chatConversations, and( + eq(chatConversations.companyId, chatActions.companyId), eq(chatConversations.id, chatActions.conversationId), + )).where(and( + sql`${chatActions.kind} = 'confirmation_response' AND ${chatActions.status} = 'processing'`, + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? or(eq(chatConversations.issueId, scope.issueId), isNull(chatConversations.id)) : undefined, + cursor ? or( + gt(chatActions.companyId, cursor.companyId), + and(eq(chatActions.companyId, cursor.companyId), gt(chatActions.id, cursor.id)), + ) : undefined, + )).orderBy(asc(chatActions.companyId), asc(chatActions.id)).limit(100); + entries.push(...rows.map(row => ({ ...row, dueAt: 0 }))); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1]; + } + throw signal?.reason ?? new Error("Confirmation recovery stopped"); + }, + check: async (entry: Entry) => { + await settleTerminalConfirmationAction(entry.id); + return null; + }, + }, canRun); + } + return { settleTerminalConfirmationAction, reconcileTerminalConfirmationActions, handleChatConfirmation, registerTerminalConfirmationWork }; } diff --git a/server/src/services/chat-confirmation-notifications.ts b/server/src/services/chat-confirmation-notifications.ts new file mode 100644 index 00000000000..e55d8c0d7ac --- /dev/null +++ b/server/src/services/chat-confirmation-notifications.ts @@ -0,0 +1,16 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ChatConfirmationScope = { companyId: string; issueId?: string }; + +export async function notifyChatConfirmationWork(tx: object, scope: ChatConfirmationScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTerminalConfirmations, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +} + +/** Interaction resolution can release a callback parked while its human input + * was pending. Register on the owning transaction before its resolution write. */ +export async function notifyChatConfirmationResolution(tx: object, interaction: ChatConfirmationScope & { kind: string }) { + if (["request_confirmation", "request_checkbox_confirmation", "suggest_tasks", "connection_intent"].includes(interaction.kind)) { + await notifyChatConfirmationWork(tx, interaction); + } +} diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 6964bd6b069..972d6789bb7 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatConfirmationWork } from "./chat-confirmation-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -13,4 +14,5 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston await notifyChatNativeProgress(tx, scope); // Admission and current chat authority also govern saved queue notices. await notifyChatInboundWakeNotices(tx, scope); + await notifyChatConfirmationWork(tx, scope); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 444165f9894..d113ff09bdf 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -59,6 +59,7 @@ export const DELIVERY_QUEUES = { chatPublications: "chat-publications", chatNativeProgress: "chat-native-progress", chatInboundWakeNotices: "chat-inbound-wake-notices", + chatTerminalConfirmations: "chat-terminal-confirmations", chatFailedRetryNotices: "chat-failed-retry-notices", chatRunMilestones: "chat-run-milestones", chatReceipts: "chat-slack-receipts", diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 363d9f2a6f5..6504b200d50 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,3 +1,4 @@ +import { notifyChatConfirmationResolution } from "./chat-confirmation-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; import { normalizeEscapedLineBreaks } from "@paperclipai/shared/validators/text"; @@ -1913,6 +1914,7 @@ async function expireStaleRequestConfirmationTarget( snapshot, }); const expireInTransaction = async (tx: Db | any) => { + await notifyChatConfirmationResolution(tx, args.row); if (args.row.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.row); const [updated] = await tx .update(issueThreadInteractions) @@ -2191,6 +2193,7 @@ export function issueThreadInteractionService( const postCommitActivityPublications: ActivityPublication[] = []; const result = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, args.issue, args.actor); + await notifyChatConfirmationResolution(tx, args.current); if (args.current.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.current); await args.mutationOptions?.beforeResolveInTransaction?.(tx); // Policy mutations and review transitions use the same issue-row lock, @@ -2475,6 +2478,7 @@ export function issueThreadInteractionService( const now = new Date(); const updated = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, args.issue, args.actor); + await notifyChatConfirmationResolution(tx, args.current); if (args.current.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.current); await args.mutationOptions?.beforeResolveInTransaction?.(tx); const issueContext = await tx @@ -2762,6 +2766,7 @@ export function issueThreadInteractionService( }) .map((candidate) => candidate.id); if (supersededIds.length > 0) { + await notifyChatConfirmationResolution(tx, { companyId: issue.companyId, issueId: issue.id, kind: "connection_intent" }); await tx .delete(toolOauthStates) .where(inArray(toolOauthStates.interactionId, supersededIds)); @@ -2860,6 +2865,7 @@ export function issueThreadInteractionService( : "expired"; const resolvedAt = now(); const updated = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, current); await mutationOptions.beforeResolveInTransaction?.(tx); const [row] = await tx .update(issueThreadInteractions) @@ -3294,6 +3300,7 @@ export function issueThreadInteractionService( const cancelled: IssueThreadInteraction[] = []; for (const row of rows) { const updated = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [resolved] = await tx @@ -3380,6 +3387,7 @@ export function issueThreadInteractionService( const expired: IssueThreadInteraction[] = []; for (const { row, replacementInteractionId } of supersededRows) { const updated = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [updatedRow] = await tx @@ -4071,6 +4079,7 @@ export function issueThreadInteractionService( const createdWakeTargets: IssueWakeTarget[] = []; await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, current); await mutationOptions.beforeResolveInTransaction?.(tx); await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); const resolvedAt = new Date(); @@ -4377,6 +4386,7 @@ export function issueThreadInteractionService( } const [updated] = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, current); await mutationOptions.beforeResolveInTransaction?.(tx); await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); const rows = await tx.update(issueThreadInteractions) @@ -4468,6 +4478,7 @@ export function issueThreadInteractionService( const expired: IssueThreadInteraction[] = []; for (const row of superseded) { const updated = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); if (row.kind === "connection_intent") { await tx @@ -4644,6 +4655,7 @@ export function issueThreadInteractionService( ); if (!sampleConfirmationRow) continue; const updatedInteractions = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, sampleConfirmationRow); if (commentRows.some(row => row.kind === "request_confirmation")) await notifyChatRunMilestones(tx, sampleConfirmationRow); await notifyToolProjectionForInteractions(tx, issue.companyId, confirmationRowIds); const updatedRows = await tx @@ -4705,6 +4717,7 @@ export function issueThreadInteractionService( for (const row of connectionIntentRows) { const [updated] = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, row); await tx .delete(toolOauthStates) .where(eq(toolOauthStates.interactionId, row.id)); @@ -4790,6 +4803,7 @@ export function issueThreadInteractionService( document, }); const updated = await db.transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, row); if (row.kind === "request_confirmation") await notifyChatRunMilestones(tx, row); const [resolved] = await tx .update(issueThreadInteractions) @@ -4853,27 +4867,29 @@ export function issueThreadInteractionService( const subject = payload?.purpose === "ai" ? "the AI account" : `the ${payload?.serviceName?.trim() || "requested"} connection`; - const [updated] = await db - .update(issueThreadInteractions) - .set({ - status: "expired", - result: { - version: 1, - outcome: "expired", - reason: `The task was reassigned before the connection request was answered, so ${subject} is still not connected. Reconnect it or choose an available connection.`, - }, - resolvedAt: now(), - updatedAt: now(), - }) - .where(and( - eq(issueThreadInteractions.id, row.id), - eq(issueThreadInteractions.status, "pending"), - )) - .returning(); - if (updated) { - await enqueueTerminalIssueInteractionChatPublications(db, hydrateInteraction(updated)); - expired.push(updated); - } + const updated = await db.transaction(async tx => { + await notifyChatConfirmationResolution(tx, { companyId: issue.companyId, issueId: issue.id, kind: "connection_intent" }); + const [resolved] = await tx + .update(issueThreadInteractions) + .set({ + status: "expired", + result: { + version: 1, + outcome: "expired", + reason: `The task was reassigned before the connection request was answered, so ${subject} is still not connected. Reconnect it or choose an available connection.`, + }, + resolvedAt: now(), + updatedAt: now(), + }) + .where(and( + eq(issueThreadInteractions.id, row.id), + eq(issueThreadInteractions.status, "pending"), + )) + .returning(); + if (resolved) await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, hydrateInteraction(resolved)); + return resolved; + }); + if (updated) expired.push(updated); } if (expired.length) await db.delete(toolOauthStates).where(inArray(toolOauthStates.interactionId, expired.map((item) => item.id))); return expired; @@ -4911,6 +4927,7 @@ export function issueThreadInteractionService( // lifecycle reflection. const updated = await db .transaction(async (tx) => { + await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); if (row.kind === "connection_intent") { await tx @@ -5002,6 +5019,7 @@ export function issueThreadInteractionService( // request must not outlive a withdrawn card. const updated = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); + await notifyChatConfirmationResolution(tx, current); if (["ask_user_questions", "request_confirmation"].includes(current.kind)) await notifyChatRunMilestones(tx, current); await resolveLinkedToolActionRequests(tx, current, { status: "cancelled", @@ -5263,6 +5281,7 @@ export function issueThreadInteractionService( const now = new Date(); const updated = await db.transaction(async (tx) => { await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); + await notifyChatConfirmationResolution(tx, current); if (["ask_user_questions", "request_confirmation"].includes(current.kind)) await notifyChatRunMilestones(tx, current); await resolveLinkedToolActionRequests(tx, current, { status: "cancelled", diff --git a/server/src/services/satisfied-connection-intents.ts b/server/src/services/satisfied-connection-intents.ts index 74503f0d098..62c29daf942 100644 --- a/server/src/services/satisfied-connection-intents.ts +++ b/server/src/services/satisfied-connection-intents.ts @@ -1,3 +1,4 @@ +import { notifyChatConfirmationResolution } from "./chat-confirmation-notifications.js"; import { enqueueTerminalIssueInteractionChatPublications } from "./chat-interaction-publications.js"; import { connectionManagementService } from "./connection-management.js"; import { chatCredentialEntryGenerations } from "./chat-channels/credentials.js"; @@ -156,6 +157,7 @@ export function satisfiedConnectionIntentService(db: Db) { if (!connection) return false; const now = new Date(); // Retire the obsolete request without inventing a human approval. + await notifyChatConfirmationResolution(tx, intent); const [retired] = await tx.update(issueThreadInteractions).set({ status: "expired", resolvedAt: now, updatedAt: now, result: { version: 1, outcome: "expired", connectionId: connection.id, reason: "An already authorized connection is available to the requesting agent." },