From 1aa3a21ac41f602977638cf0be875b01a243e69f Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 13:34:38 -0500 Subject: [PATCH] fix(server): wake inbound queue notices from committed changes (#17647) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat users receive a notice when accepted input waits for execution. > - Periodic publication maintenance scans those receipts even when nothing changes. > - Admission, cancellation, source removal, and authority changes already define when a notice can change. > - Existing action, receipt, publication, and provider-link rows preserve that work across restart. > - This PR gives ordinary inbound queue notices a commit-driven owner using the shared scheduler. > - Settled notices stop querying while unfinished batches and contention retain recovery. ## Linked Issues or Issue Description This ordinary inbound-notice slice follows merged #17895 and now targets master. Related extraction: #16187. Failed-run retry notice projection remains a separate following slice (#17649); its two shared receipt-writer notifications move forward here because ordinary input can attach to those receipt owners. **What happened?** Publication maintenance scans accepted input to project queued, not-started, and removed-source notices. Empty scans keep touching the database. Its cursor also loses PostgreSQL timestamp precision, so a full page of suppressed candidates can repeat. **Expected behavior** Committed changes should wake the notice owner. The worker should reconstruct saved candidates after restart and stop scanning when settled. It must retain a wake across drain, failed selection, contention, and a commit during reconciliation. **Steps to reproduce** Start an otherwise idle instance with chat history. Observe the ordinary inbound notice scan in periodic publication maintenance. For the cursor defect, retain a full page of denied candidates with microsecond timestamps before an eligible notice. ## What Changed - Register an ordinary inbound-notice owner with the existing delivery coordinator and scheduler. Bound each pass to 100 candidates, preserve precise continuation cursors and earlier-page contention, and reset scoped discovery after committed changes. - Notify from real admission, failed action settlement, source removal, task/tree/agent/budget/company cancellation, signed Slack Stop, and deferred queue cancellation/failure. Preserve pending-status and absent-run guards so concurrent promotion wins. - Cover task privacy/grants, project membership/visibility and current authority through their persistence owners. Keep receipt ownership, current authorization, provider policy and closed notice text in the existing chat service. - Cover shared retry-owned receipts through the existing durable authorization callback and a notifying transaction around conditional denial retirement. These small producer fixes already exist in #17649; no per-caller recovery wrappers are added. Retain its matching budget-refusal fixture correction. - Remove only ordinary notices from periodic maintenance. Preserve the separate failed-retry pass at 50 and keep the explicit combined manual API. - Preserve timestamp precision, fairness, restart, drain/resume, rollback and quiet behavior. Add two real producer transaction regressions and update the audit. Add no storage or scheduler infrastructure. ## Verification Current publication head: `41f7e2c9e351efa6cfa5ef699c0e69df7425012b`, after normal merge of master `d30c8f42b842327ee1c18c41a2efd6ba623da5cd` (#17960). The integration preserves both complete added/removed patches, all non-overlap upstream blobs/deletions, and task-before-comment lock order before the notice source lock/intent. Full binary identity is not claimed for the overlapping files. The final follow-up changes acceptance documentation only. - **202 distinct combined-master cases pass on `1720b685d9d5fb9cb4377223fdaa0a9fae2aff9c`:** 148 across the real PostgreSQL receipt concurrency and two comment route suites, plus 54 notice/FIFO/producer cases (1,598 skipped). Current server typechecking/build and module boundaries pass; all 21 owned source/test hashes are unchanged after the documentation follow-up. - The earlier CI voice-reconstruction quiet assertion failed once after 105 fetches. All 15 voice lifecycle cases pass unchanged locally; no assertions or production code were relaxed and no CI retry was used. The original failure remains in the audit, and current-head full CI must pass. - **Earlier restoration repair (head `28c898e93504318056d7a6ab2bbc5eafa6b639ae`): 126 distinct selected cases pass (54 ordinary-notice/producer cases with 1,598 skipped, plus 72 separate Retry/budget cases with 1,580 skipped).** Fix the blocking [restored input stays silent](https://github.com/paperclipai/paperclip/pull/17647#discussion_r4404007181) finding. A failed Retry's earlier receipt wake can drain before `releaseUnadmittedTaskRetryInput` restores ordinary receipts in a later transaction. The restoring transaction now awaits the existing company/task notification before its guarded write. - The new causal regression builds a durable unadmitted failed-Retry owner, uses the real input-claim owner, consumes an earlier committed wake, and proves two minutes of worker quiet with no notice. The real release transaction then restores the ordinary row, emits intent/commit settlement, and reactivates the coordinator to create its queued notice. It fails before the hook with the restored row visible but no intent. This is real PostgreSQL producer/consumer boundary proof with seeded Retry-refusal state; the separate compatibility selection exercises existing supported GitHub Retry flows. It is not a claimed complete public-request failure walkthrough. - **Earlier discovery-repair acceptance (revision `4a713edfa06b9d3dcc0e81e900006fbdb973b2d9`): 267 distinct selected cases pass:** 142 complete cancellation/app/admission consumer cases, 53 ordinary-notice cases, and 72 separate retry/budget cases. The selections skip 1,598 and 1,579 other cases. The final master integration leaves this complete patch byte-identical; the new master paths concern provider workspace/session reuse and E2E reports. - Fix the valid P2 [discovery rereads denied history](https://github.com/paperclipai/paperclip/pull/17647#discussion_r4386889825), despite the previous head's 5/5 summary. Task hints now need zero discovery SQL; company/startup discovery returns DISTINCT company/task identifiers with one query. Bounded projection still checks current authority and candidates. Three real database regressions with 250 denied actions and a second company fail before the fix and pass afterward, checking actual query and returned-row counts. Broad SQL may still inspect candidate history; no constant database-cost claim is made. - Both new producer tests fail before their fixes and pass afterward. One observes intent before a callback-owned receipt write and settlement after outer commit. The other uses real heartbeat admission/recovery and a PostgreSQL row lock to prove exact company/task intent exists before conditional denial retirement; the outside connection still sees deferred status until commit. - The reader can select a retry-owned deferred receipt for later ordinary input, and receipt resolution follows its owner. Disabling coalescing on the retry's own request does not exclude that later path. The two transferred notifications close the actual writer gaps; current authorization remains enforced by existing consumers. This is producer/transaction proof, not a claimed complete mixed-ingress walkthrough. - The moved callback exposes an old test mock that authorized against the root DB. The existing #17649 fixture correction authorizes and writes in one transaction; original requester, comment, skipped-status and budget assertions remain. Both the initial invalid lock fixture and the corrected missing-intent failures are preserved in the audit. - Updated shared dependency build, server typechecking/build and module-boundary checks pass. Because master also changed agent configuration, upgrade admission and runner contracts, the same 264 cases were rerun on combined revision `875a6ee8d211cba7f0499aace7fcea7f6880c4a2` with new isolated databases. Repeated checks are not 528 different tests. Tests under `src/__tests__` were executed separately from production typechecking. - [Current acceptance and preserved failure ledger](https://github.com/paperclipai/paperclip/blob/41f7e2c9e351efa6cfa5ef699c0e69df7425012b/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md#shared-retry-receipt-producer-gap-resolved-october-28) records current scope, exact commands, actual fixtures, source hashes and remaining limits. Earlier baseline/full-suite counts retain their original revision scope. - Server typechecking/build and module boundaries pass after the current restoration hook. The earlier combined-master compilation at `661f2f6da8c6c22cdcc25edccd57502ee6fa8c4b` retains its original scope. - **Current exact-head CI is green and Greptile is 5/5.** The old head's review error is not a pass. No full local repository matrix, live provider tenant, external alarm, physical suspension or savings claim is made. Following notice duties retain their separate owners. ### Final merge verification - Reviewed PR head remains `41f7e2c9e351efa6cfa5ef699c0e69df7425012b`: exact-head Greptile 5/5, all 55 check contexts clean, no unresolved review threads, and no Ponytail verdict present. - One Runner shard initially failed descriptor suspension and a five-second OpenCode fixture deadline. The same package passed on the preceding head. The first local attempt was invalid because required Rust fixture binaries were absent; after building them, all eight relevant cases pass with unchanged timeouts. One failed-job retry then passes 1,956 cases (four skipped). No source/assertion/time-budget change or second CI retry was used. - Current master `0e365768f032a472807b5720da8431f92d2f07ce` adds manual monitor and task-drain fixes. Its routes file overlaps but merges cleanly. Local normal-merge validation commit `2720b79c769cb609ad4d392512331e67b910ff3f` preserves both complete added/removed patches and every unrelated master blob/deletion. This is not raw binary identity for overlapping files. - **156 distinct cases pass on that exact combined tree:** all 102 monitor/drain/policy cases plus 54 notice cases (1,598 skipped). Types and build pass. An initial competing-cold-check run failed setup hooks (38 passed, two hook failures, 62 skipped); the same four suites pass alone with all original hook/test timeouts and assertions intact. The failed run remains recorded, not counted as full acceptance. - This merge proof changes no published source or review head. The protected pinned-head merge guard recomputes both patches and requires the exact tested master/tree/head, successful validation receipts and unchanged live merge gates. Previous acceptance counts retain their original revision scope; no local full-repository or deployed-sleep claim is made. Master then advanced to `75754ffb7af2b7ee7dce1e915a145faa82a8fb2a` with #17976 (five UI/e2e/map files only). The entire `server` and `packages` trees are identical to the validated composition. Both complete patches and all newly landed UI files are preserved in the final clean merge tree; the protected gate checks these exact source-root hashes and changed paths. No additional UI-test result is claimed. ## Risks - Missing a supported producer could delay a notice until restart. Tests cover real service and API producers; durable rows remain authoritative. Notifications cover supported application writers in the owning process, not arbitrary external SQL writers. - Privacy invalidation can affect descendants, so those changes use company scope. Ordinary admission and publication changes use task scope. Settled history causes no recurring scan. - Provider output remains subject to current authority and receipt ownership. A cancelled request without a visible queued notice does not acquire a new failure message. - This establishes worker quiet, not host or database suspension. Cloud alarms and conservative sleep gates do not change. Other maintenance remains in place. - Remaining stacked slices still need their own integration, review, and CI. This slice merges only under the user’s explicit campaign authorization after current-head gates pass. ## Model Used OpenAI Codex, GPT-6 family, with repository inspection and code execution tools. The exact serving variant and context-window size are not exposed in this session. ## 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 - [x] All Paperclip CI gates are green - [x] 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 --- ...2026-10-24-recurring-server-sleep-stack.md | 16 + ...-10-25-inbound-wake-notices-sleep-audit.md | 595 +++++++++++++ .../app-private-hostname-gate.test.ts | 5 + .../chat-channels.integration.test.ts | 796 +++++++++++++++++- server/src/app.ts | 2 + .../modules/wake-queue/adapters/postgres.ts | 3 + .../adapters/queued-comment-postgres.ts | 2 + server/src/routes/issues.ts | 21 +- server/src/services/chat-channels.ts | 8 +- .../chat-channels/failed-run-retry.ts | 19 +- .../chat-channels/inbound/wakeup-notices.ts | 94 ++- .../services/chat-channels/inbound/wakeups.ts | 5 + .../chat-channels/slack/session-stop.ts | 2 + .../chat-channels/task-retry-input.ts | 2 + .../chat-inbound-wake-notice-notifications.ts | 7 + .../services/chat-inbound-wake-notice-work.ts | 44 + .../chat-run-milestone-notifications.ts | 3 + server/src/services/companies.ts | 4 + .../services/delivery-work-notifications.ts | 1 + server/src/services/heartbeat/run-control.ts | 52 +- server/src/services/issue-tree-control.ts | 48 +- server/src/services/issues.ts | 13 + server/src/services/projects.ts | 13 +- 23 files changed, 1649 insertions(+), 106 deletions(-) create mode 100644 doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md create mode 100644 server/src/services/chat-inbound-wake-notice-notifications.ts create mode 100644 server/src/services/chat-inbound-wake-notice-work.ts 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 a756578e9c1..f9bebd56ecd 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2188,3 +2188,19 @@ base-publication hook. The hook fixes a reproduced commit-order wake gap; its 16 cases at `bcf2f38ff5`; final typecheck passes too. Remote checks/review are pending. Parent #17606 now has passing checks and Greptile 5/5 on `ed76256b97`; all three review threads are resolved. This is not completion of the broader stack. + +## Current gate update: native progress and inbound notices (2026-10-25) + +#17639 at `6c40618ec1bb098e1c7fe4098133d1de71d287ee` now has all checks +passing (CI run 40218349023), a fresh Greptile 5/5 review of that commit, and +zero review threads. The circular signal-payload and over-wide atomic-event +selection failures were repaired, not bypassed by rerunning unchanged failures. +The PR remains unmerged. + +Slice 30 now registers its scoped notice worker and removes normal inbound +notices from periodic publication maintenance. Focused acceptance passes 42 +cases. Its full verification and new PR are still pending. See the dated inbound +wake notice audit for the producer inventory, regressions and fixture corrections. +This update does not clear the older Grok/runtime-discovery CI failures, voice +session recovery review, or inbound voice cleanup ownership findings, and it does +not shrink the remaining original checklist. diff --git a/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md b/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md new file mode 100644 index 00000000000..b16cf52c0c3 --- /dev/null +++ b/doc/plans/2026-10-25-inbound-wake-notices-sleep-audit.md @@ -0,0 +1,595 @@ +# Inbound wake notices: sleep migration audit + +Parent: #17639 at `34df8b0d0116a14d414a70d29f64a16c5672a335`. +Branch: `codex/sleep-30-inbound-wake-notices`. +Status: owner/producer audit in progress. Production still polls this duty. + +## One reason and its adjacent owners + +This slice owns the safe pre-run notices for admitted inbound chat messages: +queued, not started, and removed. The existing owner is +`chat-channels/inbound/wakeup-notices.ts`. These are projections of admission +receipts, not a second scheduler for executing the admitted task. + +`chat-channels.ts::processPublicationMaintenance` calls the notice scan on every +chat maintenance pass. The explicit/manual `processPendingPublications` API also +calls it when maintenance is enabled. Production queued-publication delivery +already has a maintenance-free path, which should stay independent. + +The current `enqueueInboundWakeupPublications` first invokes +`enqueueFailedChatRetryPublications`. That projects failed-run retry receipts, +with a separate authorization contract in `failed-run-retry.ts`. Separate this +adjacent reason rather than claiming both migrated by moving a compound call. +Terminal confirmation actions, interaction continuations, and GitHub maintenance +are additional separate reasons and remain pending. + +## Durable contract + +Existing `chat_actions`, `agent_wakeup_requests`, `chat_deliveries`, source +comments, publications and message links own the obligation. No new table is +needed. A normal inbound action must be processed or failed and have the exact +admission receipt. A coalesced receipt must resolve to a matching canonical owner. +Queued requires deferred execution without a run; a failed/cancelled/skipped +receipt must not be presented as a queue wait. + +A not-started/removed follow-up can retire only the exact previously visible +queued notice. The published provider message ID and outbound message link are +part of that proof. Deleting one coalesced source must not erase a surviving +batch's response. Promotion, questions, selected final answers and replacement +provider lanes retain precedence. Text remains the existing closed phrases: +no input text, internal scheduler failure, queue position, ETA or model content. + +`currentInboundWakeNotice` owns current issue, endpoint, identity and receipt +checks. It takes the existing nonblocking locks to avoid inversion with ingress +and scheduling. Preserve this authorization at projection and again at delivery. +A denied candidate is settled until a relevant mutation; lock contention is +unresolved work and needs a retry. The current catch treats both as zero inserted, +so a count-only return is insufficient once the permanent poll disappears. + +## Selector and lifecycle work required + +The current owner stores a process-global Date/UUID cursor between bounded calls. +That cursor belongs neither to a company nor to an issue and loses PostgreSQL +microsecond precision. Do not reuse it as a scoped worker cursor. Use stable SQL +precision and a complete per-pass continuation, advance past suppressed rows, +and preserve a concrete obligation when a page or lock remains unfinished. + +Use the existing scoped-deadline worker. Startup reconstructs saved work; scoped +commits refresh only matching work; empty or settled history disarms. Retry only +contention, failed reconciliation and unfinished batches. Changes during a pass, +first writes during drain, failed startup, resume and shutdown require the same +shared lifecycle contract as earlier slices. + +## Producer audit: do not remove the poll yet + +- `inbound/wakeups.ts` settles actual accepted actions. It signals lifecycle + projection for processed actions, but failed actions can also change notices. + Preserve the distinction between durable admission and a callback returning. +- `heartbeat/queue.ts` creates deferred receipts, coalesces requests, promotes + owners and changes terminal receipt status. Existing milestone signals cover + some run transitions, not every receipt-only mutation. Inspect the existing + writer transaction owners before adding a notification contract. +- `execution-wait.ts` can persist skipped admission without creating a run. + Determine which transitions can create or retire an eligible notice. +- `issue-tree-control.ts` cancels queued/deferred receipts without runs. + Its current notification is for terminal workspace cleanup, not these notices. +- `chat-channels/slack/session-stop.ts` can cancel a queued receipt. A signal for + the stop-action worker is not automatically a notice wake. +- `companies.ts` cancels queued/deferred/claimed receipts during company state + changes. Preserve company visibility and disablement behavior. +- Source-comment removal and later receipt promotion can retire a visible notice; + audit `issues.ts`, message lifecycle owners and queued-comment control paths. +- Publication success/message-link persistence can make a retirement eligible. + The existing issue-scoped publication notification is a possible shared + invalidation boundary; ordinary lease/claim bookkeeping must remain narrow. +- Endpoint, connection, resource, identity, membership and task visibility can + restore authority. Reuse the existing eligibility contract only after checking + these notice predicates; lifecycle projection coverage is not proof by itself. +- Remaining production receipt writers include heartbeat run control/lifecycle, + retries, recovery, task retry, human-directed work, native status decisions, + continuations and provider-capacity retry. Classify supported notice-producing + paths explicitly. A search hit alone is neither coverage nor a required wake. + +Notifications must precede the write and follow the existing outer transaction. +Do not create an external call or a recovery object at every route. Do not wake +for unchanged health checks or claim renewal. Process-local signals retain the +single active writer model; external writers require their own integration. + +## Acceptance + +Existing integration fixtures `deferredChatQueueFixture` and its pre-run FIFO +cases exercise coalescing, restart, successor reuse, exact final/question lanes, +revoked reach, source deletion, identity/session changes, canonical owner IDs, +and receipt-lock contention. Establish that baseline, then add real-producer +worker tests with no manual projection call or periodic backstop. + +New cases must prove configured-idle/history quiet, first durable admission, +outer commit/rollback, receipt-only cancellation, source removal after promotion, +visible-queue receipt ordering, restoration of authority, suppressed full pages, +scoping/fairness, lock retries, drain/resume, concurrent changes and shutdown. +Do not turn queued work into periodic notice polling after its notice is saved. + +This audit changes no runtime behavior or Cloud eligibility. Exact implementation, +producer coverage and acceptance remain required before a migration PR is ready. + +## Groundwork evidence + +The existing 22 FIFO/notice integration cases pass in +`/tmp/sleep30-inbound-notice-baseline.log`. A new regression then reproduces the +cursor defect: 50 authority-rejected actions share a PostgreSQL microsecond +timestamp before one valid admitted action; two bounded passes both return zero. +Evidence: `/tmp/sleep30-notice-cursor-before.log`. + +Preserving the timestamp as PostgreSQL text repairs that defect without changing +eligibility or batching. All 23 notice cases pass in +`/tmp/sleep30-notice-cursor-fixed.log` on a fresh isolated external PostgreSQL +fixture database. Production typecheck passes in +`/tmp/sleep30-cursor-typecheck.log`. The new test deletes its artificial candidates +so they cannot starve later tests. Scoped worker ownership and complete producer +coverage remain unimplemented; this is groundwork, not a polling-removal claim. + +## Scoped owner and first producers (2026-10-25) + +Parent notification repair `3447973f6f` is merged into this local branch. +The owner now exposes scoped candidate reconstruction and normal-notice-only +reconciliation. Each scoped pass has its own microsecond-precise keyset cursor, +advances past suppressed pages, and reports actual lock contention separately +from settled authorization denial. The manual API retains its bounded cursor +and combined failed-run retry behavior. The new scoped-deadline worker is +implemented but **not yet wired into production startup**; the periodic backstop +remains until all producer and lifecycle coverage is complete. + +The existing lifecycle/authority notification boundary now also invalidates +normal inbound notices. Real chat admission wakes the worker after a quiet +startup, and settled/published history becomes quiet again. Removing that +notification makes the acceptance case fail (no queued publication), proving +the case exercises the producer rather than a manual scan. + +Task-tree cancellation and source-comment tombstoning each reproduced a missed +retirement wake. Tree cancellation now registers intent inside its transaction; +caller-owned transactions retain outer commit/rollback semantics. Tombstoning +reads and locks the still-live comment's company/task binding before registering +intent and updating it; already deleted/missing comments remain no-ops. These +changes preserve existing notice text and current authorization. Tests deliver +the queued publication with maintenance disabled, perform real service mutations +inside a rolled-back outer transaction and then a committed transaction, and +observe the existing provider message being edited once. Two minutes of fake +idle time cause no further root select/transaction calls. + +29 scoped/manual/admission/retirement integration cases passed in +`/tmp/sleep30-notice-producer-acceptance.log`. Producer regression evidence: +`/tmp/sleep30-notice-worker-before-producer.log`, +`/tmp/sleep30-notice-tree-cancel-before.log`, and +`/tmp/sleep30-notice-source-removal-before.log`. Initial worker-test failures +were missing `canRun` in its coordinator fixture and a wait that checked only +the timer while a sweep was still running; the corrected wait also requires +idle-work settlement. These fixture failures are not production defects. +Server typecheck passed in `/tmp/sleep30-notice-producer-typecheck.log`. + +Still required: remaining receipt/cancellation writers (including budget, agent, +company and Slack stop paths), admission-failure settlement, authority changes +not already covered by shared notifications, batching/scoping/lifecycle tests, +production registration, and removal of only this duty from the maintenance +loop. Fresh PR creation and all checks remain outstanding for slice 30. + +## Receipt audit refinement (in progress) + +Do not mechanically notify every `agent_wakeup_requests` write. The normal +notice predicate requires an exact `inbound_wakeup` action, matching durable +receipt/canonical owner, and current source/destination authority. A first queued +notice requires an unadmitted deferred owner. A not-started retirement requires +both receipt and owner to have no run. Removed-source retirement instead requires +the exact visible queued message and a soft-deleted source; it can remain eligible +after surviving coalesced input promotes. + +- First durable chat receipt insertion is followed by the existing + `inbound/wakeups.ts` processed-action settlement, now carrying notice intent. + Failed action settlement is still an explicit gap to address. Do not assume + the earlier admission transaction alone establishes a processed action. +- `execution-wait.ts` increments counters on an already-skipped receipt or + inserts a new skipped receipt. The counter update does not change notice + eligibility; first-admission terminal settlement belongs to the action owner. +- `heartbeat/queue.ts` periodic `updatedAt` touches (queued-comment interrupt and + stranded-queue inspection) and `run-control.ts` retry cursor touches must not + wake notice reconciliation. `executionWait` explanatory-message-only updates + do not change the closed notice text or eligibility. +- `explicit-native-continuation.ts` and `fresh-workspace-continuation.ts` change + the task-retry workspace strategy only; no notice wake follows from that field. +- `human-directed-work.ts::retainBacklogHumanAssignment` inserts a completed + board-provenance receipt, not a deferred inbound-chat admission. +- `setWakeupStatus` call sites found in heartbeat queue/control/completion, + recovery and retries operate on a run's receipt. Recheck any exceptional + run-less caller before adding a broad signal: a receipt with a run cannot + produce not-started, and source tombstoning owns removed-source activation. +- Explicit cancellation paths still needing integration/tests include + `run-control.ts::cancelPendingWakeupsForBudgetScope`, `cancelPendingWakeupsForAgentsInternal`, company archive cascade, + Slack session stop, and queue replay paths that decline an unadmitted receipt. +- `issues.ts` reassignment cancels user receipts in the issue-update transaction; + it currently signals lifecycle only for unhide, so that signal is not coverage + of reassignment. Current-authority restoration and mutation ordering need an + explicit test. +- Dot/Muse follow-up consumption changes deferred receipt payloads and can bind + them to an already-running turn. Audit their provider-lane precedence rather + than emitting a false not-started notice for consumed input. +- Task-retry and failed-run-retry generations have separate notice policy. Check + whether normal inbound receipts may reference their canonical owners before + deciding which writes belong to this reason or the next failed-retry slice. + +The full chat integration suite was launched at local branch `c579148977` on a +fresh `paperclip_sleep30_full_chat_20261025_test` database, with output at +`/tmp/sleep30-full-chat-acceptance.log`; its result must be read before claiming +acceptance. Typecheck after merging the second parent repair passed in +`/tmp/sleep30-after-parent-typecheck.log`. + +## Producer and lifecycle completion (2026-10-25) + +The full chat suite at `c579148977` finished successfully: 1,458 tests in +`/tmp/sleep30-full-chat-acceptance.log`. This was the baseline before the +cancellation and privacy changes below, not acceptance of later edits. + +Real service regressions reproduced stale visible queue notices after agent +cancellation, budget cancellation, and company archival. Signals now belong to +those owners' transactions, as well as Slack Stop and deferred queue cancellation +and failure. Agent cancellation now rechecks pending status and absence of a run +at update time, so it cannot cancel a receipt promoted after candidate selection. +Caller-owned transactions preserve outer commit and rollback. Company lifecycle +intentionally owns its root transaction; its test respects that contract. + +Task status/assignment changes and privacy changes can restore eligibility. +Task privacy can affect descendants, so it invalidates the company scope. Explicit +task grants and private-project membership use their existing persistence owners; +the project-member helper owns a small transaction that also nests under callers. +Project visibility changes use the existing project transaction. Shared chat +identity, endpoint, resource and membership signals already cover the other +current-authority predicates. Ordinary health refreshes and receipt cursor touches +do not signal this worker. Run-bound settlement and Dot/Muse consumption do not +create a not-started notice; their existing response/progress owners retain that +policy. Failed-run retry notices remain a separate owner. + +Failed inbound-action settlement and repaired delivery acceptance now invalidate +normal notices when a valid durable request exists. Preparing actions whose +underlying deliveries were rejected cannot satisfy the notice predicate and do +not acquire a notice obligation. Publication completion already signals the task +inside the transaction that saves the provider receipt and message link. A new +race test holds provider I/O, removes the source, observes the notice worker +settle before the receipt exists, and proves that the eventual receipt commit +reactivates the removed-source edit. + +The app now registers `chatInboundWakeNoticeWork` with the existing coordinator. +`processPublicationMaintenance` retains only the separate failed-run retry notice +pass, plus its other unmigrated duties. The explicit/manual combined +`enqueueInboundWakeupPublications` API remains available. + +Focused acceptance at the current source passes 42 cases in +`/tmp/sleep30-wired-acceptance.log`. These include actual admission after idle, +scoped/manual cursor pagination, denied history, receipt-lock contention, +transaction rollback and commit, queue rejection/failure, agent/budget/company +cancellation, privacy restoration through task grants and project membership, +signed Slack Stop, late provider receipt, restart deduplication, writes after +shutdown, first write and failed selection during drain, and a commit arriving +while an empty selection returns. Quiet assertions observe root SQL entry points +after advancing fake time by two minutes. The lifecycle tests do not prove host +or database suspension, and no Cloud sleep-safety blocker is relaxed. + +Preserved failures and fixture corrections: +- `/tmp/sleep30-cancellation-before.log`: agent cancellation missed its wake; + an attempted agent reassignment fixture was rejected by the existing external + channel binding rule. The unsupported fixture was removed, not bypassed. +- `/tmp/sleep30-budget-company-before.log`: budget cancellation missed its wake; + company lifecycle correctly rejected a caller transaction. The root-company + reproduction in `/tmp/sleep30-company-root-before.log` then showed its missing + wake. All five initial retirement cases passed after the repair. +- `/tmp/sleep30-notice-privacy-before.log`: opening a private task did not wake + its saved notice. The repair passes. Initial grant tests used an unlinked + external principal as a user; corrected fixtures establish a linked board user + through the normal chat setup before admission. +- The first signed Stop fixture had an invalid Slack channel ID and a verifying, + rather than qualified active, endpoint. Correcting both makes the signed event + reach cancellation. `/tmp/sleep30-signed-stop-qualified.log` passes. + +Workspace typecheck/build, full acceptance after startup wiring, PR publication, +and latest-head CI/review are still required before this slice is ready. + + +The first workspace-wide typecheck and build at `d4f7426bb4` found a missing +required limit in the newly separated failed-retry maintenance call (TS2554). +The corrected call passes 50, preserving the old combined wrapper's default. +The full chat run at that revision was deliberately interrupted (exit 130) after +this compiler evidence; it is not acceptance. Its log remains at +`/tmp/sleep30-wired-full-chat.log`. Fresh verification is required after repair. + + +## Bounded candidate passes after review + +At `87f5c827b0`, workspace typecheck and build passed, and the full chat plus +startup-gate selection passed all 1,486 tests in +`/tmp/sleep30-final-full-chat.log`. CI subsequently passed every code gate; +Greptile found a real fairness defect (comment 4371894496): the reconciliation +limit counted inserted notices rather than candidates authorized, so a denied +history could monopolize one pass. + +The new regression fails before repair in `/tmp/sleep30-fairness-before.log`: +a requested 50-candidate pass opens more than 50 candidate transactions. The +repair returns a precise continuation cursor after at most the requested page +size. The notice worker retains that cursor between scheduler passes. It also +retains contention from earlier pages and, after reaching the end, revisits it +from the beginning rather than silently discarding locked work. A committed +scope invalidation reconstructs the entry with a fresh cursor; no shared +scheduler or authorization policy changes are needed. + +All 44 selected integration cases pass in +`/tmp/sleep30-fairness-acceptance.log`, including a 150-row denied backlog, +an unrelated task completing between bounded passes, contention on the first +page surviving a later page, and a real privacy-restoration write while a page +is in flight. The latter revisits the formerly denied prefix, publishes exactly +once, and then performs no root SQL for two minutes of fake idle time. Manual +pagination remains separate and retains PostgreSQL microsecond precision. +Server typecheck passed after the cursor contract change. Full latest-head +remote checks and review must be revalidated after pushing the repair. + +The next owner audit must explicitly trace ordinary receipts coalesced into a +failed-run retry receipt: `findExistingDeferredWake` does not itself exclude a +retry owner. `allowRunCoalescing: false` on the retry's own admission is not +proof that a later ordinary admission cannot select it. In particular, inspect +the retry authorization-denial write before assuming it can only affect retry +notices. This is an open producer-audit question, not a verified exemption. + + +## Current-parent baseline and producer follow-up (October 28) + +The existing slice is prepared with a normal merge of the verified cancellation +parent `09c0b8dc92f589d0a3c53716581e43863a0bb819`. Its original published head is +`ca8f2d17c64fc7cecff4526817d1ac6e1d479ff4`, with own base +`fffc8d0c46fbbeb3da4a23a36e638fefd3b8fe3d`. All original added/removed lines and +remaining diff context are preserved. Differences in the patch representation +are blob IDs, line coordinates, and one inherited decision-target import that +was renamed in the current parent. The automatic merge is clean; the renamed +import and every unrelated parent path are retained. + +The existing-patch baseline passes **190 distinct selected cases**: all 142 +cancellation, app-wiring, admission-application and queued-comment application +cases across four complete suites, plus 48 real chat cases selected for inbound +notices/FIFO/deferred-admission behavior. The other 1,598 chat cases were skipped, +not passed. Server typechecking, build, and module boundaries pass. Logs and +zero-status receipts are in `/tmp/sleep30-current-parent-*`; the exact selected +commands and isolated fixtures are in +`/tmp/sleep30-current-parent-baseline-command.json`. + +This is baseline evidence, not a ready-to-merge claim. The mixed ordinary/retry +receipt producer question above remains open in this checkpoint. Source tracing +confirms that a later ordinary admission can select an existing retry-owned +deferred receipt: the reader has no retry-owner exclusion, and disabling +coalescing on the retry's own request does not exclude later ordinary requests. +The following failed-retry slice already adds ordinary-notice notifications to +retry authorization/admission and conditional denial retirement. Determine which +minimal writer coverage must move into this owner before publication; verify its +transaction behavior rather than claiming those receipts cannot be shared. +Final producer disposition, updated acceptance, actual landed-parent integration, +and fresh exact-head remote gates remain required. No complete local repository +matrix, live provider, host/database suspension, or savings claim is made. + + +## Shared retry-receipt producer gap resolved (October 28) + +The current preparation adopts verified cancellation parent +`60ba892ae9fc5faaf8cd0833694b4a66c13323c3`. Its complete original inbound-notice +baseline patch remains identical through that parent update. The following +producer repair closes the explicitly open question above before this owner +is published. + +`findExistingDeferredWake` can select a deferred retry receipt for a later +ordinary admission with matching task, conversation, and actor. The retry's own +`allowRunCoalescing: false` controls its initial admission, not that later read. +Ordinary receipt resolution follows `coalescedIntoWakeupRequestId` to its owner; +owner retirement can therefore change an ordinary notice. Do not exempt retry +writers merely because their initial admission disables coalescing. + +Two already-implemented writer fixes are moved forward from the following +failed-retry slice (#17649, source `b47169855cc712e029df7e5c4c34f7b6fa04b9d4`): + +- The durable retry authorization callback registers ordinary-notice work after + validation and before its caller writes the receipt. This keeps the contract + central for both real heartbeat admission and supported callback-owned writes. +- Retry authorization denial retires the exact unadmitted receipt in a transaction + that registers its company/task notice intent before the existing guarded update. + Pending-status, absent-run, and request-correlation conditions are preserved. + +No new worker, table, scheduler, polling fallback, or per-caller recovery wrapper +is added. These production hunks match the already-existing following slice. +Its remaining failed-retry projection and lifecycle/lease producers keep their +separate ownership. Ordinary notice authorization does not read native lease or +coordinator retry eligibility, so those later predicates are not copied here. + +Two new real producer regressions fail before repair and pass afterward: + +1. A supported callback authorizes and writes a real deferred retry receipt in + its own transaction. There was no notice intent before the write. The fixed + callback emits intent before the write and settlement after outer commit. +2. Real heartbeat admission creates the retry receipt. After actual retry receipt + reconciliation, denial reaches its conditional retirement. A second PostgreSQL + transaction holds that receipt row: the fixed writer emits exact company/task + intent while the outside connection still sees deferred status, remains unsettled + while blocked, then publishes settlement only after the failed-status write commits. + +These verify actual producers and transaction timing. They do not claim a complete +mixed ordinary/retry ingress walkthrough; the admission reader, receipt resolver, +scoped consumer tests, and the shared owner writes establish the audited boundary. +The first lock fixture acquired the row too early and blocked admission itself; +that invalid before-fix attempt is preserved at +`/tmp/sleep30-retry-owner-red.log`. The corrected missing-intent failure is +`/tmp/sleep30-retry-owner-actual-red.log`. The callback failure is +`/tmp/sleep30-retry-admission-red.log`; both pass together in +`/tmp/sleep30-retry-producers-green.log`. + +Moving the callback contract also exposes the existing budget-refusal test mock's +root `request.authorize(db)` call. The real writer passes its transaction. The +local still-budget-blocked case failed with no skipped receipt and a contended +retry classification (`/tmp/sleep30-retry-budget-red.log`). Its correction is the +same transaction fixture fix already present in #17649: authorize and persist +inside one transaction. All requester, comment, skipped-status, and budget checks +remain; production budget behavior is unchanged. The reusable deferred-retry +fixture is copied from that following slice for the two new producer cases. +When reconciling #17649, retain one identical helper and do not duplicate it. + +### Final current-parent acceptance + +**264 distinct selected cases pass**, with separate newly created PostgreSQL +fixtures for reconstruction, broader retries, and cancellation consumers: + +- **142 complete consumer cases** across cancellation/run control, app admission + wiring, wake admission use cases, and queued-comment admission use cases. +- **50 selected ordinary-notice cases**, including both new producer regressions; + the other 1,598 chat cases were skipped. +- **72 different retry cases**, covering the full newer-input budget-hold family, + current GitHub Retry, and persisted Retry authority loss; the other 1,576 chat + cases were skipped. These selected names do not overlap the 50-case group. + +Server typechecking, build, and module-boundary checks pass. The tests under +`src/__tests__` were executed separately from production typechecking. The exact +commands, fixtures, logs, and zero exit receipts are in +`/tmp/sleep30-final-acceptance-command.json` and `/tmp/sleep30-final-*`. +`/tmp/sleep30-final-tested-source-sha256.json` records the 20 owned source/test +hashes held stable throughout the final validation. + +The earlier 190-case baseline and older full-suite counts retain their stated +revision scope and are not added to 264. This completes the named producer audit +for this owner; actual landed-parent reconciliation and fresh exact-head CI/review +remain publication and merge gates. There is no full local repository matrix, +live provider tenant, external-alarm, physical suspension, or savings claim. + + +### Acceptance on actual landed master + +The final normal ancestry merge is `875a6ee8d211cba7f0499aace7fcea7f6880c4a2`, +based on actual master `b14e1ecaf85ce6eafa05125f1b539e3ac7f32022`, after +cancellation #17895 merged. The complete 22-path notice patch is binary-identical +across that integration and all 58 unrelated master paths are preserved. + +Master added agent YAML configuration/shared types, bounded upgrade admission, +and runner human-input contracts. Patch identity alone does not establish their +combined runtime behavior. Updated shared dependencies, server typechecking, +build, and module boundaries pass. The same **264 distinct cases** then pass +again on this actual combined revision: 142 complete consumer cases, 50 selected +notice cases, and 72 separate retry/budget cases, using new isolated databases. +The selected skipped counts remain 1,598 and 1,576 respectively. These repeated +runs do not represent 528 different tests. + +Exact commands, revision, and actual fixtures are in +`/tmp/sleep30-landed-acceptance-command.json`; logs and zero exit receipts are +`/tmp/sleep30-landed-{shared-build,types,build,boundaries,consumers,notices,retries}.*`. +All 20 owned source/test hashes remain unchanged from the recorded acceptance +manifest. The only subsequent publication edit adds this documentation. Fresh +remote current-head CI and review remain required before protected merge. + + +### Review repair: discover scopes without rereading action pages + +Greptile's current-head 5/5 summary also contained a valid P2: +[discovery rereads denied history](https://github.com/paperclipai/paperclip/pull/17647#discussion_r4386889825). +A task notification already contains its scope, but discovery reread every page +of denied actions before bounded authorization could begin. Company/startup +selection likewise transferred all candidate actions merely to deduplicate tasks. + +Task notifications now return their supplied scope without SQL. The subsequent +bounded reconciliation remains responsible for checking current candidates and +authority, including an already exhausted scope. Company and startup discovery +use one SQL DISTINCT projection of company/task identifiers. The same candidate +query supplies both discovery and bounded selection, preserving its owner, +company, UUID, source-removal, and publication predicates. Broad discovery may +still inspect candidate history within PostgreSQL; this is not a constant-cost +query or a new recurring scan. + +Three real PostgreSQL regressions seed 250 denied candidates and a second company. +They fail before the fix (two discovery queries instead of zero/one), then pass: +zero discovery queries/rows for task hints, one query with only distinct scope +rows for company/startup. The existing exhausted-history assertions now exercise +company discovery; task hints deliberately need no database existence check. +Cancellation, projection budgets and precise cursor behavior remain unchanged. + +The revised matrix passes **267 distinct cases**: 142 complete consumer cases, +53 selected ordinary-notice cases (1,598 skipped), and 72 separate retry cases +(1,579 skipped). Fresh isolated databases keep retained obligations independent. +Server typechecking, build, and module boundaries pass. Exact commands/fixtures +are in `/tmp/sleep30-discovery-final-acceptance-command.json`; before/after evidence +is `/tmp/sleep30-discovery-{red,green}.log`, and acceptance logs/status receipts +are `/tmp/sleep30-discovery-final-{consumers,notices,retries}.*`. +The earlier 264-case acceptance retains its original revision scope. None of +these runs establishes full local repository acceptance or deployed sleep. + + +### Review repair: restore ordinary Retry input in its notifying transaction + +The fresh review of `661f2f6da8c6c22cdcc25edccd57502ee6fa8c4b` reports +[restored input stays silent](https://github.com/paperclipai/paperclip/pull/17647#discussion_r4404007181), +P1/4 out of 5. Earlier retry admission and denial notifications do not cover +`releaseUnadmittedTaskRetryInput`: reconciliation invokes this owner after the +failed action settles, and its separate transaction restores coalesced ordinary +receipts to deferred status. A consumed earlier wake cannot cover that later +commit. The earlier writer audit missed this restoration transaction. + +The restoration owner now awaits the existing company/task notice intent after +its action/issue/receipt guards and before updating the ordinary receipts. The +same transaction restores input and retires its unadmitted task Retry intent. +No guard, source authority, table, scheduler or per-caller wrapper is added. + +The causal regression seeds an unadmitted failed Retry owner and uses the real +input-claim transaction on supported ordinary chat input. It starts the actual +notice coordinator, consumes an earlier committed scope wake, and observes no +notice and two virtual minutes without select/transaction calls. Calling the +real release owner then exposes the restored row. Before the fix, the test fails +because this transaction emitted no intent. After the fix, intent/settlement and +the queued publication appear, and the coordinator settles without a deadline. +This is real PostgreSQL producer/consumer boundary proof with seeded durable +Retry-refusal state, not a complete public Retry request failure walkthrough. +The separate existing GitHub Retry compatibility cases exercise supported flows. + +Current restoration acceptance passes **126 distinct selected cases**: 54 +ordinary-notice/FIFO/producer cases (1,598 skipped) and 72 separate retry/budget +cases (1,580 skipped), using fresh independent databases. Server typechecking, +build and module boundaries pass. The prior 267-case discovery matrix retains +its original revision scope; unchanged complete consumer suites are not counted +as freshly rerun for this two-line producer hook. +Commands/fixtures: `/tmp/sleep30-restore-input-final-acceptance-command.json`. +The valid missing-intent failure remains in `/tmp/sleep30-restore-input-red.log`; +final results/exit receipts are `/tmp/sleep30-restore-input-final-{notices,retries}.*` +and `/tmp/sleep30-restore-input-{types,build,boundaries}.*`. +Fresh current-head remote CI and Greptile 5/5 remain mandatory. No full local +repository acceptance, live provider wake or physical suspension is claimed. + + +### Reconcile durable comment writes on current master + +After the restoration repair earned exact-head Greptile 5/5, master landed +#17960 at `d30c8f42b842327ee1c18c41a2efd6ba623da5cd`. Its 62 changed paths +include durable comment receipts and task-before-comment mutation locking. +The normal merge `1720b685d9d5fb9cb4377223fdaa0a9fae2aff9c` resolves two conflicts: +both route imports remain; soft deletion takes the upstream task lock before +our source-comment row lock and awaited notice intent. Receipt completion and +task deletion therefore retain their common lock order. + +Both branches' complete added/removed patches are preserved in both directions; +only diff blob hashes/hunk metadata are excluded from the comparison. All +non-overlap upstream blobs and deleted paths match master exactly. This is not +a claim of full-context or binary identity for the two overlapping files. +`/tmp/sleep30-durable-comment-master-composition-proof.json` records the proof, +including the corrected check for an upstream-deleted UI hook. + +Combined revision acceptance passes **202 distinct cases**: all 148 cases in +three durable receipt/reopen/wakeup suites, including the real embedded +PostgreSQL receipt concurrency suite, plus 54 notice/FIFO/producer cases on a +fresh external PostgreSQL fixture (1,598 skipped). Server typechecking, build +and module boundaries pass. Logs/exit receipts use +`/tmp/sleep30-durable-comment-{receipt-acceptance,notices,types,build,boundaries}.*`. +The earlier 126-case restoration and 267-case discovery records retain their +original revision scopes. A following documentation-only commit leaves the +21 owned server source/test hashes unchanged, as recorded in +`/tmp/sleep30-durable-comment-tested-source-sha256.json`. + +The earlier `661f2f6da8` CI run also failed the existing voice report reconstruction +quiet assertion after 105 fetches (one late transaction). All 15 voice lifecycle +cases pass unchanged on a fresh local database. No assertion or production code +was relaxed and no CI retry was used. This is not a proven root cause; the full +current-head CI must pass. The original failure and local result remain in +`/tmp/sleep30-discovery-ci-server3-direct.log` and `/tmp/sleep30-voice-ci-baseline.log`. +Fresh CI and review are required for the final publication, including when an +older head already earned 5/5. No deployed sleep claim is added. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 77cf0d1e5c8..29b46cddee7 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -79,6 +79,7 @@ describe("createChatReconciliationCoordinator", () => { expect(maintenance).not.toContain("chatChannels.voice."); expect(maintenance).not.toContain("enqueueSafeNativeChatProgress"); expect(source).toContain("nativeChatProgressWork(db).register(deliveryWork,"); + expect(source).toContain("chatInboundWakeNoticeWork(chatChannels).register(deliveryWork,"); expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,"); expect(source).toContain("chatChannels.registerTeamsConsentExpiryWork(deliveryWork,"); const teams = readFileSync(new URL("../services/chat-channels/microsoft-teams/file-delivery.ts", import.meta.url), "utf8"); @@ -87,6 +88,10 @@ describe("createChatReconciliationCoordinator", () => { expect(teams).not.toContain("selectDistinct"); const chat = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8"); expect(chat).not.toContain("scheduleTeamsFileMaintenance"); + const publicationMaintenance = chat.slice(chat.indexOf("async function processPublicationMaintenance("), + chat.indexOf("async function schedulePendingPublications(")); + expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications("); + expect(publicationMaintenance).toContain("await enqueueFailedChatRetryPublications("); }); it("separates live voice recovery from the remaining periodic maintenance", () => { diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index fff734e5b36..29d1b0db701 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,8 +1,11 @@ +import { issueTreeControlService } from "../services/issue-tree-control.js"; +import { chatInboundWakeNoticeWork } from "../services/chat-inbound-wake-notice-work.js"; import { nativeChatProgressWork } from "../services/chat-native-progress-work.js"; import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "../middleware/auth.js"; import { cloudTenantPrimaryCompanyId } from "../services/cloud-instance.js"; import { accessService } from "../services/access.js"; import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; +import { claimTaskRetryInput, releaseUnadmittedTaskRetryInput } from "../services/chat-channels/task-retry-input.js"; import { restoreNativeChatReviewPresentationInTransaction } from "../services/native-runtime/native-chat-review-presentation.js"; import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js"; import { deleteCompany } from "../services/company-deletion.js"; @@ -192,6 +195,9 @@ import { } from "../services/chat-teams-personal-recipient.js"; import * as discordQuestionForms from "../services/chat-discord-question-forms.js"; import { issueService } from "../services/issues.js"; +import { companyService } from "../services/companies.js"; +import { ensureProjectAccessMember } from "../services/projects.js"; +import { createPostgresWakeQueueAdapter } from "../modules/wake-queue/adapters/postgres.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { getExternalChannelBindingSummary } from "../services/chat-channel-binding.js"; import { PaperclipRunnerToolAuthority } from "../services/native-runtime/paperclip-runner-tool-authority.js"; @@ -61692,6 +61698,126 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); }); + async function deferredFailedRetryNoticeFixture(wakeOnDemand = true) { + const context = await failedChatRetryFixture("telegram", undefined, true); + const companyId = context.fixture.companyId, issueId = context.issue.id, leaseId = randomUUID(); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, + nativePhase: "terminal_failure", runnerProfileJson: {}, + resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: { + schema: "paperclip.workspace-restore-recovery.v1", runId: context.runId, leaseIds: [leaseId], + } }, + }).where(eq(heartbeatRuns.id, context.runId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: context.runId, phase: "terminal_failure" }); + await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId, + heartbeatRunId: context.runId, provider: "daytona", providerLeaseId: randomUUID(), + status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "retain_on_failure", + metadata: { remoteCwd: "/home/daytona/paperclip-workspace", pendingCleanupInFlight: false }, + }); + await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", status: "resolved", ownerType: "system", + cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), + nextAction: "Verify and preserve the retained source before continuing.", + evidence: { runId: context.runId, automaticRecovery: { replay: "blocked" }, + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }, + historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: context.runId, leaseIds: [leaseId] }, + systemRecovery: { ...createSystemRecovery(context.runId), state: "exhausted", attempts: 2, retriesUsed: 1, + attemptId: randomUUID(), firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000, + preservationOperationIds: [randomUUID()] }, + }, + }); + const retry = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user", + actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(), + prepareChatRetry: context.service.prepareFailedChatRunRetry }); + await db.update(agents).set({ adapterType: "paperclip_runner", status: "running", + runtimeConfig: { heartbeat: { wakeOnDemand, maxConcurrentRuns: 1 } }, + }).where(eq(agents.id, context.fixture.assignedAgentId)); + await db.insert(heartbeatRuns).values({ companyId, agentId: context.fixture.assignedAgentId, + status: "running", startedAt: new Date() }); + context.wakeup.mockImplementation(heartbeatService(db).wakeup); + return { ...context, retry, admit: () => context.service.processFailedChatRunRetry(retry.chatActionId!) }; + } + + it("registers ordinary-notice work through the retry admission transaction", async () => { + const context = await deferredFailedRetryNoticeFixture(); + const events: string[] = []; + let eventsBeforeWrite: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatInboundWakeNotices, (event, scopes) => { + if ([...scopes ?? []].some(scope => scope !== "*" && JSON.parse(scope).companyId === context.fixture.companyId)) events.push(event); + }); + context.wakeup.mockImplementationOnce(async (_agentId, options) => { + const request = options!.durableChatRequest!; + await db.transaction(async tx => { + events.length = 0; + await request.authorize(tx as unknown as Parameters[0]); + eventsBeforeWrite = [...events]; + await tx.insert(agentWakeupRequests).values({ + id: request.id, companyId: context.fixture.companyId, agentId: request.agentId, + source: options!.source, triggerDetail: options!.triggerDetail, reason: options!.reason, + requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, + requestedAt: request.requestedAt, idempotencyKey: request.idempotencyKey, + status: "deferred_issue_execution", payload: { ...options!.payload, + taskRetryRequestId: context.retry.requestId, _paperclipWakeContext: options!.contextSnapshot }, + }); + }); + return null; + }); + try { + await expect(context.admit()).resolves.toMatchObject({ status: "deferred" }); + expect(eventsBeforeWrite).toContain("intent"); + expect(eventsBeforeWrite).not.toContain("settled"); + expect(events).toContain("settled"); + expect((await readAgentWakeupRequestsById(context.retry.chatActionId!))[0]).toMatchObject({ status: "deferred_issue_execution", runId: null }); + } finally { unsubscribe(); await context.service.shutdown(); } + }); + + it("registers ordinary-notice intent before a retry-owned receipt is retired", async () => { + const context = await deferredFailedRetryNoticeFixture(); + let release = () => {}; + let holding: Promise | undefined; + let retiring: ReturnType | undefined; + let unsubscribe = () => {}; + const events: string[] = []; + const signalledScopes: string[] = []; + try { + expect(await context.admit()).toMatchObject({ status: "deferred" }); + await reconcileTaskChatRetryReceipts(db); + context.wakeup.mockClear(); + let locked!: () => void; + const acquired = new Promise(resolve => { locked = resolve; }); + const held = new Promise(resolve => { release = resolve; }); + context.wakeup.mockImplementationOnce(async () => { + holding = db.transaction(async tx => { + await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, context.retry.chatActionId!)).for("update"); + locked(); + await held; + }); + await acquired; + throw Object.assign(new Error("denied at resume"), { + details: { code: "chat_failed_run_retry_not_authorized" }, + }); + }); + unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatInboundWakeNotices, (event, scopes) => { + if ([...scopes ?? []].some(scope => scope !== "*" && JSON.parse(scope).companyId === context.fixture.companyId)) { + events.push(event); signalledScopes.push(...scopes ?? []); + } + }); + retiring = context.admit(); + void retiring.catch(() => undefined); + await vi.waitFor(() => expect(context.wakeup).toHaveBeenCalledOnce()); + await vi.waitFor(() => expect(events).toContain("intent"), { timeout: 2000 }); + expect(signalledScopes.map(scope => JSON.parse(scope))).toContainEqual({ companyId: context.fixture.companyId, issueId: context.issue.id }); + expect(events).not.toContain("settled"); + expect((await readAgentWakeupRequestsById(context.retry.chatActionId!))[0]).toMatchObject({ status: "deferred_issue_execution", runId: null }); + release(); await holding; + await expect(retiring).resolves.toMatchObject({ status: "failed" }); + expect((await readAgentWakeupRequestsById(context.retry.chatActionId!))[0]).toMatchObject({ status: "failed", runId: null }); + expect(events).toContain("settled"); + } finally { + release(); await holding; await retiring?.catch(() => undefined); + unsubscribe(); await context.service.shutdown(); + } + }); + async function failedChatRetryFixture( provider: "slack" | "telegram" | "github", externalActorId?: string, @@ -62632,13 +62758,15 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { context.wakeup.mockImplementation(heartbeatService(db).wakeup); } else context.wakeup.mockImplementation(async (agentId, opts) => { const request = opts.durableChatRequest!; - await request.authorize(db); - await db.insert(agentWakeupRequests).values({ id: request.id, companyId, - agentId, source: opts.source!, triggerDetail: opts.triggerDetail, reason: "budget.blocked", - requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, - requestedAt: request.requestedAt, idempotencyKey: request.idempotencyKey, - payload: { ...opts.payload, executionWait: { scopeType: "agent", scopeId: agentId } }, - status: "skipped", finishedAt: new Date() }); + await db.transaction(async tx => { + await request.authorize(tx); + await tx.insert(agentWakeupRequests).values({ id: request.id, companyId, + agentId, source: opts.source!, triggerDetail: opts.triggerDetail, reason: "budget.blocked", + requestedByActorType: request.requestedByActorType, requestedByActorId: request.requestedByActorId, + requestedAt: request.requestedAt, idempotencyKey: request.idempotencyKey, + payload: { ...opts.payload, executionWait: { scopeType: "agent", scopeId: agentId } }, + status: "skipped", finishedAt: new Date() }); + }); return null; }); if (realBudgetIngress) { @@ -73085,10 +73213,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(committed.providerRuntime.slackFilePublicationAttempts).toBe(2); }); - async function deferredChatQueueFixture(admissionFails = false) { + async function deferredChatQueueFixture(admissionFails = false, beforeAdmission?: (service: ChatChannelService) => Promise, linkedBoardUser = false) { const fixture = await seedCompany(); let ownerId: string | null = null; const configured = await configuredSlackEndpoint(fixture, { + linkedBoardUser, wakeup: async (agentId, opts) => { if (admissionFails) throw new Error("PRIVATE admission failure token=do-not-publish"); @@ -73144,8 +73273,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }, }); const thread = makeThread({ - channelId: "C-DEFERRED-NOTICE", - id: "slack:C-DEFERRED-NOTICE:1999000.1", + channelId: "CDEFERREDNOTICE", + id: "slack:CDEFERREDNOTICE:1999000.1", }); const send = async (id = "1999000.1") => deliverMessage({ @@ -73155,10 +73284,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { message: makeMessage({ id, text: "@maya exact queued request", + userId: linkedBoardUser ? "UBOARD" : undefined, mentioned: id === "1999000.1", }), trigger: id === "1999000.1" ? "mention" : "subscribed_message", }); + await beforeAdmission?.(configured.service); await send(); const [conversation] = await readChatConversationsByEndpointId(configured.endpoint.id); const [action] = await db @@ -73205,6 +73336,651 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("wakes pre-run FIFO notices from real admission after quiet and becomes quiet again", async () => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + const errors: unknown[] = []; + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + 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 { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready; + await settle(); await quiet(); + }); + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + commentId: context.action.payload.commentId, + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + await quiet(); + await coordinator!.stop(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); await quiet(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(errors).toEqual([]); + } finally { + await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); + } + }); + + it("reconstructs inbound admission committed after the notice worker stops", async () => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + const errors: unknown[] = []; + try { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready; + await coordinator.stop(); + }); + 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(); } + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); } + }); + + it.each(["task visibility", "project membership", "task grant"] as const)("wakes a saved inbound notice when %s restores privacy access", async kind => { + const context = await deferredChatQueueFixture(false, undefined, true); + await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.conversation.issueId, + createdByUserId: null, responsibleUserId: null, assigneeUserId: null, + }).where(eq(issues.id, context.conversation.issueId)); + 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(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + if (kind === "task visibility") { + await issueService(db).update(context.conversation.issueId, { visibility: "open" }); + } else if (kind === "project membership") { + const [project] = await db.insert(projects).values({ companyId: context.fixture.companyId, + name: "Private notice project", visibility: "private", + }).returning(); + await db.update(issues).set({ projectId: project.id }).where(eq(issues.id, context.conversation.issueId)); + await ensureProjectAccessMember(db, { companyId: context.fixture.companyId, projectId: project.id, + subjectType: "user", subjectId: String(context.action.payload.requestedByActorId), + }); + } else { + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { req.actor = { ...boardActor(context.fixture.companyId), isInstanceAdmin: true }; next(); }); + app.use("/api", issueRoutes(db, createStorageService().storage)); app.use(errorHandler); + await request(app).post(`/api/issues/${context.conversation.issueId}/access-grants`).send({ + subjectType: "user", subjectId: String(context.action.payload.requestedByActorId), + }).expect(201); + } + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it.each(["startup", "first admission", "failed selection"] as const)("retains inbound notices through %s during drain without passive SQL", async kind => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + let canRun = kind !== "startup"; + const errors: unknown[] = []; + try { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const selection = kind === "failed selection" + ? vi.spyOn(channel, "loadInboundWakeNoticeScopes").mockRejectedValueOnce(new Error("temporary notice selection outage")) + : undefined; + try { await chatInboundWakeNoticeWork(channel).register(coordinator, () => canRun).ready; } + finally { selection?.mockRestore(); } + canRun = false; + }); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + 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(); } + canRun = true; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toHaveLength(kind === "failed selection" ? 1 : 0); + } finally { await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); } + }); + + it("retains admission committed while an empty inbound-notice selection is returning", async () => { + let coordinator: ReturnType | undefined; + let service: ChatChannelService | undefined; + let selection: ReturnType | undefined; + let ready: Promise | undefined; + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + const errors: unknown[] = []; + try { + const context = await deferredChatQueueFixture(false, async channel => { + service = channel; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const load = channel.loadInboundWakeNoticeScopes.bind(channel); + selection = vi.spyOn(channel, "loadInboundWakeNoticeScopes").mockImplementationOnce(async (...args) => { + const rows = await load(...args); held = true; await pending; return rows; + }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + ready = chatInboundWakeNoticeWork(channel).register(coordinator, () => true).ready; + await vi.waitFor(() => expect(held).toBe(true)); + }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + release(); await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { release(); await ready; selection?.mockRestore(); await coordinator?.stop(); vi.useRealTimers(); await service?.shutdown(); } + }); + + it.each(["tree cancellation", "source removal", "agent cancellation", "budget cancellation", "company archive", "queue rejection", "queue failure"] as const)("wakes a visible pre-run FIFO retirement only after %s commits", async change => { + const context = await deferredChatQueueFixture(); + const releaseRunId = randomUUID(); + if (change === "queue rejection" || change === "queue failure") { + await db.insert(heartbeatRuns).values({ id: releaseRunId, companyId: context.fixture.companyId, + agentId: context.fixture.assignedAgentId, status: "succeeded", finishedAt: new Date(), + contextSnapshot: { issueId: context.conversation.issueId }, + }); + } + 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(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const notices = () => readChatPublicationsByEndpointId(context.endpoint.id); + const expectedText = change === "source removal" ? "This queued message was removed." + : "This follow-up was not started. Open the task in Paperclip for details."; + const retire = async (database: typeof db) => { + if (change === "source removal") return issueService(database).tombstoneComment(String(context.action.payload.commentId), { + actorType: "user", userId: String(context.action.payload.requestedByActorId), + }); + if (change === "agent cancellation") return heartbeatService(database).cancelInvocationsForAgents( + [context.fixture.assignedAgentId], "PRIVATE cancellation", + ); + if (change === "budget cancellation") return heartbeatService(database).cancelBudgetScopeWork({ + companyId: context.fixture.companyId, scopeType: "company", scopeId: context.fixture.companyId, + }); + if (change === "company archive") return companyService(database).update(context.fixture.companyId, { status: "archived" }); + if (change === "queue rejection" || change === "queue failure") { + const adapter = createPostgresWakeQueueAdapter(database, { + resolveResponsibleUserId: async () => "owner-user", + getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }), + resolveSessionBeforeForWakeup: async () => null, + }); + return adapter.withIssueExecutionLock({ companyId: context.fixture.companyId, runId: releaseRunId, now: new Date() }, async (_locked, ports) => { + const target = { companyId: context.fixture.companyId, wakeId: context.action.id, now: new Date() }; + if (change === "queue rejection") await ports.transaction.cancelDeferredWake({ ...target, reason: "PRIVATE rejection" }); + else await ports.transaction.failDeferredWake(target); + return { outcome: { kind: "released" }, postCommitEffects: [] }; + }); + } + return issueTreeControlService(database).cancelUnclaimedWakeupsForTree( + context.fixture.companyId, context.conversation.issueId, "PRIVATE cancellation", + ); + }; + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + await context.service.processPendingPublications(100, { maintenance: false }); + await settle(); + expect(await notices()).toEqual([expect.objectContaining({ state: "published" })]); + // Company lifecycle owns its root transaction and intentionally rejects + // caller transactions. The other service producers support nesting. + if (change !== "company archive") { + await expect(db.transaction(async tx => { + await retire(tx as unknown as typeof db); + await vi.advanceTimersByTimeAsync(1000); + expect(await notices()).toHaveLength(1); + throw new Error("outer cancellation rollback"); + })).rejects.toThrow("outer cancellation rollback"); + await settle(); expect(await notices()).toHaveLength(1); + } + await retire(db); + await settle(); + expect(await notices()).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: expectedText, progressState: "failed" }, + })])); + expect(await notices()).toHaveLength(2); + await context.service.processPendingPublications(100, { maintenance: false }); + await settle(); + expect(context.runtime.endpoints.get(context.endpoint.id)!.edits).toEqual([expect.objectContaining({ + text: expectedText, + })]); + 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(); } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("rechecks source retirement after a queued notice's late provider receipt commits", async () => { + const context = await deferredChatQueueFixture(); + 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 }); + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + let publishing: Promise | undefined; + const provider = context.runtime.endpoints.get(context.endpoint.id)!; + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + provider.postHook = async () => { held = true; await pending; }; + publishing = context.service.processPendingPublications(100, { maintenance: false }); + await vi.waitFor(() => expect(held).toBe(true)); + await issueService(db).tombstoneComment(String(context.action.payload.commentId), { + actorType: "user", userId: "owner-user", + }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + release(); await publishing; provider.postHook = undefined; + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This queued message was removed.", progressState: "failed" }, + })])); + await context.service.processPendingPublications(100, { maintenance: false }); + await settle(); + expect(provider.edits).toEqual([expect.objectContaining({ text: "This queued message was removed." })]); + expect(errors).toEqual([]); + } finally { release(); await publishing; provider.postHook = undefined; await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it("wakes inbound notice retirement from a signed Slack Stop without a maintenance scan", async () => { + const context = await deferredChatQueueFixture(false, undefined, true); + await qualifySetupRoundTrip(context.service, context.endpoint.id, "UBOARD"); + await context.service.test(context.endpoint.id, "owner-user"); + 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(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + try { + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); await context.service.processPendingPublications(100, { maintenance: false }); await settle(); + const body = JSON.stringify({ type: "event_callback", team_id: "T-PAPERCLIP", event_id: "EvSleepQueueStop", + event: { type: "agent_session_stopped", channel: "CDEFERREDNOTICE", thread_ts: "1999000.1", + event_ts: `${Math.floor(Date.now() / 1000)}.999999`, streaming_message_ts: [], user: "UBOARD" }, + }); + const response = await context.service.handleWebhook(context.endpoint.publicId, "slack", signedSlackWebhookRequest({ + body, contentType: "application/json", url: `https://paperclip.example/api/chat-webhooks/${context.endpoint.publicId}/slack`, + })); + expect(response.status).toBe(202); + await settle(); + expect(await db.select({ status: chatActions.status, result: chatActions.result }).from(chatActions) + .where(and(eq(chatActions.endpointId, context.endpoint.id), eq(chatActions.kind, "slack_session_stop")))) + .toEqual([expect.objectContaining({ status: "processed", result: expect.objectContaining({ code: "slack_session_stop_cancelled" }) })]); + expect(await readAgentWakeupRequestsById(context.action.id)).toEqual([expect.objectContaining({ status: "cancelled" })]); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This follow-up was not started. Open the task in Paperclip for details.", progressState: "failed" }, + })])); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); } + }); + + it.each(["manual", "scoped"] as const)("advances pre-run FIFO selection past a suppressed page with microsecond timestamps (%s)", async mode => { + const context = await deferredChatQueueFixture(); + const rejectedIds = Array.from({ length: 50 }, () => randomUUID()); + try { + const [receipt] = await readAgentWakeupRequestsById(context.action.id); + const before = new Date(context.action.createdAt.getTime() - 1000).toISOString(); + await db.insert(chatActions).values(rejectedIds.map(id => ({ + ...context.action, id, providerActionId: `inbound_wakeup:${id}`, + payload: { ...context.action.payload, commentId: randomUUID() }, + createdAt: sql`${before}::timestamptz + interval '0.000123 seconds'`, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ + ...receipt, id, idempotencyKey: `chat-cursor-fixture:${id}`, + }))); + // Rejected current authority must not keep the same full page at the + // head of every bounded pass and starve the later admitted message. + if (mode === "manual") { + const first = await context.service.enqueueInboundWakeupPublications(50); + const second = await context.service.enqueueInboundWakeupPublications(50); + expect(first + second).toBe(1); + } else { + const scope = { companyId: context.fixture.companyId, issueId: context.conversation.issueId }; + const transaction = vi.spyOn(db, "transaction"); + try { + const first = await context.service.reconcileInboundWakeNotices({ scope, limit: 50 }); + expect(transaction).toHaveBeenCalledTimes(50); + expect(first).toMatchObject({ inserted: 0, contended: false, nextCursor: expect.any(Object) }); + transaction.mockClear(); + expect(await context.service.reconcileInboundWakeNotices({ scope, limit: 50, cursor: first.nextCursor })) + .toEqual({ inserted: 1, contended: false, nextCursor: null }); + expect(transaction).toHaveBeenCalledTimes(1); + } finally { transaction.mockRestore(); } + } + const rows = await readChatPublicationsByEndpointId(context.endpoint.id); + expect(rows).toEqual([expect.objectContaining({ + commentId: context.action.payload.commentId, + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + } finally { + await context.service.shutdown(); + await db.delete(chatActions).where(inArray(chatActions.id, [...rejectedIds, context.action.id])); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, [...rejectedIds, context.action.id])); + } + }); + + it.each(["denied", "contended"] as const)("yields a %s inbound notice backlog between bounded passes without losing work", async mode => { + const context = await deferredChatQueueFixture(); + const other = await deferredChatQueueFixture(); + const rejectedIds = Array.from({ length: 150 }, () => randomUUID()); + let release = () => {}; + let holder: Promise | undefined; + try { + const [receipt] = await readAgentWakeupRequestsById(context.action.id); + const at = new Date(context.action.createdAt.getTime() + (mode === "denied" ? -1000 : 1000)).toISOString(); + await db.insert(chatActions).values(rejectedIds.map(id => ({ + ...context.action, id, providerActionId: `inbound_wakeup:${id}`, + payload: { ...context.action.payload, commentId: randomUUID() }, + createdAt: sql`${at}::timestamptz + interval '0.000123 seconds'`, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ + ...receipt, id, idempotencyKey: `chat-fairness-fixture:${id}`, + }))); + if (mode === "contended") { + let locked!: () => void; + const ready = new Promise(resolve => { locked = resolve; }); + const hold = new Promise(resolve => { release = resolve; }); + holder = db.transaction(async tx => { + await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, context.action.id)).for("update"); + locked(); await hold; + }); + await ready; + } + const worker = chatInboundWakeNoticeWork(context.service); + const [entry] = await worker.load({ companyId: context.fixture.companyId, issueId: context.conversation.issueId }); + const [otherEntry] = await worker.load({ companyId: other.fixture.companyId, issueId: other.conversation.issueId }); + const transaction = vi.spyOn(db, "transaction"); + try { + let remaining = await worker.check(entry!); + expect(transaction).toHaveBeenCalledTimes(100); + expect(remaining).toMatchObject({ cursor: expect.any(Object), contended: mode === "contended" }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]); + // Another scope can finish while the first still has a saved cursor. + expect(await worker.check(otherEntry!)).toBeNull(); + expect(await readChatPublicationsByEndpointId(other.endpoint.id)).toHaveLength(1); + transaction.mockClear(); + remaining = await worker.check(remaining!); + expect(transaction).toHaveBeenCalledTimes(51); + if (mode === "contended") { + expect(remaining).toMatchObject({ cursor: null }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]); + release(); await holder; + remaining = await worker.check(remaining!); + expect(remaining).not.toBeNull(); + remaining = await worker.check(remaining!); + } + expect(remaining).toBeNull(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + commentId: context.action.payload.commentId, + payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + } finally { transaction.mockRestore(); } + } finally { + release(); await holder; + await context.service.shutdown(); await other.service.shutdown(); + await db.delete(chatActions).where(inArray(chatActions.id, [...rejectedIds, context.action.id])); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, [...rejectedIds, context.action.id])); + } + }); + + it("restarts an inbound notice cursor when privacy is restored during a bounded pass", async () => { + const context = await deferredChatQueueFixture(false, undefined, true); + const rejectedIds = Array.from({ length: 100 }, () => randomUUID()); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let held = false, release!: () => void; + const pending = new Promise(resolve => { release = resolve; }); + let ready: Promise | undefined; + const reconcile = context.service.reconcileInboundWakeNotices.bind(context.service); + const pass = vi.spyOn(context.service, "reconcileInboundWakeNotices").mockImplementationOnce(async input => { + const result = await reconcile(input); + expect(result).toMatchObject({ inserted: 0, nextCursor: expect.any(Object) }); + held = true; await pending; return result; + }); + try { + await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.conversation.issueId, + createdByUserId: null, responsibleUserId: null, assigneeUserId: null, + }).where(eq(issues.id, context.conversation.issueId)); + const [receipt] = await readAgentWakeupRequestsById(context.action.id); + const at = new Date(context.action.createdAt.getTime() + 1000).toISOString(); + await db.insert(chatActions).values(rejectedIds.map(id => ({ + ...context.action, id, providerActionId: `inbound_wakeup:${id}`, + payload: { ...context.action.payload, commentId: randomUUID() }, createdAt: new Date(at), + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ + ...receipt, id, idempotencyKey: `chat-cursor-reset:${id}`, + }))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + ready = chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await vi.waitFor(() => expect(held).toBe(true), { timeout: 10_000 }); + await issueService(db).update(context.conversation.issueId, { visibility: "open" }); + release(); await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + 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(); } + expect(errors).toEqual([]); + } finally { + release(); await ready; await coordinator.stop(); pass.mockRestore(); vi.useRealTimers(); + await context.service.shutdown(); + await db.delete(chatActions).where(inArray(chatActions.id, [...rejectedIds, context.action.id])); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, [...rejectedIds, context.action.id])); + } + }); + + it.each(["task", "company", "startup"] as const)("bounds inbound notice scope discovery with denied history (%s)", async mode => { + const context = await deferredChatQueueFixture(); + const other = await deferredChatQueueFixture(); + const rejectedIds = Array.from({ length: 250 }, () => randomUUID()); + try { + const [receipt] = await readAgentWakeupRequestsById(context.action.id); + await db.insert(chatActions).values(rejectedIds.map(id => ({ + ...context.action, id, providerActionId: `inbound_wakeup:${id}`, + payload: { ...context.action.payload, commentId: randomUUID() }, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ + ...receipt, id, idempotencyKey: `chat-discovery-budget:${id}`, + }))); + const scope = { companyId: context.fixture.companyId, issueId: context.conversation.issueId }; + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 1, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { chatActions, agentWakeupRequests }, capture); + let scopes: Array<{ companyId: string; issueId: string }>; + try { + scopes = await context.service.loadInboundWakeNoticeScopes( + mode === "task" ? scope : mode === "company" ? { companyId: scope.companyId } : undefined, + ); + } finally { capture.flush(); } + if (mode === "startup") { + expect(scopes).toEqual(expect.arrayContaining([scope, { + companyId: other.fixture.companyId, issueId: other.conversation.issueId, + }])); + } else { + expect(scopes).toEqual([scope]); + } + expect(events.at(-1)).toMatchObject({ startedQueries: mode === "task" ? 0 : 1, + completedQueries: mode === "task" ? 0 : 1, inFlightAtEnd: 0 }); + expect(events.filter(event => event.event === "paperclip.db_response_capture.entry.v1") + .reduce((count, event) => count + Number(event.rows), 0)).toBe(mode === "task" ? 0 : scopes.length); + } finally { + await context.service.shutdown(); await other.service.shutdown(); + await db.delete(chatActions).where(inArray(chatActions.id, [...rejectedIds, context.action.id])); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, [...rejectedIds, context.action.id])); + } + }); + + it("reactivates restored ordinary input after an earlier Retry notice wake is consumed", async () => { + const context = await deferredChatQueueFixture(); + const retryId = randomUUID(), intentId = randomUUID(); + const scope = { companyId: context.fixture.companyId, issueId: context.conversation.issueId }; + const errors: unknown[] = [], events: string[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let unsubscribe = () => {}; + try { + const [receipt] = await readAgentWakeupRequestsById(context.action.id); + await db.insert(chatActions).values({ ...context.action, id: retryId, + kind: "failed_run_retry", status: "failed", providerActionId: `task-retry-restore:${retryId}`, + payload: { issueId: scope.issueId, taskRetryIntentId: intentId }, + }); + await db.insert(agentWakeupRequests).values({ ...receipt, id: retryId, status: "failed", + idempotencyKey: `task-retry-restore:${retryId}`, + }); + await db.transaction(tx => claimTaskRetryInput(tx, { companyId: scope.companyId, + retryActionId: retryId, sourceActionIds: [context.action.id] })); + expect((await readAgentWakeupRequestsById(context.action.id))[0]).toMatchObject({ + status: "coalesced", runId: null, payload: { coalescedIntoWakeupRequestId: retryId }, + }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + await chatInboundWakeNoticeWork(context.service).register(coordinator, () => true).ready; + await db.transaction(tx => signalDatabaseWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices, JSON.stringify(scope))); + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]); + 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(); } + unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatInboundWakeNotices, (event, scopes) => { + if ([...scopes ?? []].some(value => value !== "*" && JSON.parse(value).companyId === scope.companyId)) events.push(event); + }); + await releaseUnadmittedTaskRetryInput(db, { companyId: scope.companyId, retryActionId: retryId }); + expect((await readAgentWakeupRequestsById(context.action.id))[0]).toMatchObject({ status: "deferred_issue_execution", runId: null }); + expect(events).toContain("intent"); expect(events).toContain("settled"); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + commentId: context.action.payload.commentId, payload: { text: "Your follow-up is queued.", progressState: "queued" }, + })]); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { + unsubscribe(); await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown(); + await db.delete(chatActions).where(eq(chatActions.id, retryId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, retryId)); + } + }); + + it("isolates pre-run FIFO reconciliation scopes and exhausts published history", async () => { + const first = await deferredChatQueueFixture(); + const second = await deferredChatQueueFixture(); + try { + const scope = { companyId: first.fixture.companyId, issueId: first.conversation.issueId }; + expect(await first.service.loadInboundWakeNoticeScopes({ companyId: scope.companyId })).toEqual([scope]); + expect(await first.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 1, contended: false, nextCursor: null }); + expect(await readChatPublicationsByEndpointId(second.endpoint.id)).toEqual([]); + expect(await first.service.loadInboundWakeNoticeScopes({ companyId: scope.companyId })).toEqual([]); + expect(await first.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false, nextCursor: null }); + expect(await second.service.reconcileInboundWakeNotices({ + scope: { companyId: second.fixture.companyId, issueId: second.conversation.issueId }, + })).toEqual({ inserted: 1, contended: false, nextCursor: null }); + } finally { + await first.service.shutdown(); await second.service.shutdown(); + } + }); + + it("retains a contended pre-run FIFO projection and settles cancelled input without a visible notice", async () => { + const context = await deferredChatQueueFixture(); + let release!: () => void; + let locked!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { locked = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, context.action.id)).for("update"); + locked(); await hold; + await tx.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, context.action.id)); + }); + try { + await ready; + const scope = { companyId: context.fixture.companyId, issueId: context.conversation.issueId }; + expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: true, nextCursor: null }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]); + release(); await holder; + expect(await context.service.reconcileInboundWakeNotices({ scope })).toEqual({ inserted: 0, contended: false, nextCursor: null }); + expect(await context.service.loadInboundWakeNoticeScopes({ companyId: scope.companyId })).toEqual([]); + } finally { + release(); await holder; await context.service.shutdown(); + } + }); + it("keeps one durable pre-run FIFO notice across coalescing and restart, then reuses only its successor lane", async () => { const context = await deferredChatQueueFixture(); let restarted: ChatChannelService | null = null; diff --git a/server/src/app.ts b/server/src/app.ts index cdba74fb00f..ffb2e212678 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -196,6 +196,7 @@ import { toolAccessService } from "./services/tool-access.js"; import { chatChannelService } from "./services/chat-channels.js"; import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js"; import { nativeChatProgressWork } from "./services/chat-native-progress-work.js"; +import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; import { createCoalescedAsyncTrigger, @@ -1288,6 +1289,7 @@ export async function createApp( chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatRunMilestoneWork(db, opts.authPublicBaseUrl).register( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index 90c24c75a57..f76cbacb844 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js"; import { isAcknowledgedNativeReassignmentStop, isAcknowledgedNativeStop } from "../../../services/acknowledged-native-stop.js"; @@ -313,6 +314,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }, async cancelDeferredWake({ companyId, wakeId, reason, now }) { + await notifyChatInboundWakeNotices(tx, { companyId }); const rows = await tx .update(agentWakeupRequests) .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now }) @@ -344,6 +346,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru }, async failDeferredWake({ companyId, wakeId, now }) { + await notifyChatInboundWakeNotices(tx, { companyId }); const rows = await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index e2c0e2f740f..242f9089a3d 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../../services/chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../../../services/chat-run-milestone-notifications.js"; import { and, eq, inArray } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -94,6 +95,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu }, async cancelWake({ wakeId, reason, now }) { + await notifyChatInboundWakeNotices(tx, { companyId, issueId }); await tx .update(agentWakeupRequests) .set({ status: "cancelled", finishedAt: now, error: reason, updatedAt: now }) diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 9ab7951bd46..245f2314dd6 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js"; import { completeIssueCommentReceipt } from "../services/issue-comment-receipt.js"; import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; import { parseObject } from "../adapters/utils.js"; @@ -13508,14 +13509,18 @@ export function issueRoutes( const subjectId = req.body.subjectId as string; await assertIssueGrantSubjectExists(issue.companyId, subjectType, subjectId); const actor = getActorInfo(req); - const result = await db.transaction((tx) => ensureActiveIssueAccessGrant(tx, { - issueId: managedIssue.id, - subjectType, - subjectId, - source: "explicit", - grantedByUserId: actor.actorType === "user" ? actor.actorId : null, - grantedByAgentId: actor.agentId, - })); + const result = await db.transaction(async (tx) => { + // Grants on an ancestor can restore current chat authority on descendants. + await notifyChatInboundWakeNotices(tx, { companyId: issue.companyId }); + return ensureActiveIssueAccessGrant(tx, { + issueId: managedIssue.id, + subjectType, + subjectId, + source: "explicit", + grantedByUserId: actor.actorType === "user" ? actor.actorId : null, + grantedByAgentId: actor.agentId, + }); + }); if (result.created) { await logActivity(db, { companyId: issue.companyId, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 8589877dce4..f08bc42cba4 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1258,7 +1258,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { withCredentialMutationLease, teamsFileRetryAfter, isShuttingDown: () => shuttingDown, get teamsFileUploadRequest() { return options.teamsFileUploadRequest; }, }); - const { enqueueInboundWakeupPublications, authorizeInboundWakePublication } = createInboundWakeNotices(db, { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange, isShuttingDown: () => shuttingDown }); + const { enqueueInboundWakeupPublications, loadInboundWakeNoticeScopes, reconcileInboundWakeNotices, authorizeInboundWakePublication } = createInboundWakeNotices(db, { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange, isShuttingDown: () => shuttingDown }); async function authorizeRetainedChatSourcePublication( tx: DbOrTransaction, @@ -3348,7 +3348,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPublicationMaintenance(limit = 25) { scheduleGitHubMaintenance(); - await enqueueInboundWakeupPublications(); + // Normal inbound notices are commit-driven. Failed-run retries retain their + // separate maintenance owner until its own producer audit is complete. + await enqueueFailedChatRetryPublications(50); await reconcileTerminalConfirmationActions(limit); await processPendingInteractionWakeups(limit); } @@ -3551,6 +3553,8 @@ return { reconcileProviderRuntimes, processPendingPublications, enqueueInboundWakeupPublications, + loadInboundWakeNoticeScopes, + reconcileInboundWakeNotices, schedulePendingPublications, processPendingDeliveries, processQueuedDeliveries: () => processPendingDeliveries(25, undefined, false), diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index cd5cf7c0bf5..26b53a0e1e8 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { safeChatWakeAuthorityReason, type ChatWakeAuthorityReason } from "./inbound/wakeup-authorization-error.js"; import { adapterExecutionControls } from "../adapter-execution-control.js"; import { admitExplicitNativeContinuation } from "../explicit-native-continuation.js"; @@ -1871,6 +1872,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti typeof claimed.payload.taskRetryIntentId === "string" ? claimed.payload.taskRetryIntentId : undefined, true, // The durable wake callback validates admission; later phases recheck containment. ); + await notifyChatInboundWakeNotices(tx, { companyId: scope.companyId, issueId: scope.issueId }); }, onAdmitted: async (tx, receipt) => { if (!source.taskRetryInput) return; @@ -1915,12 +1917,17 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti receipt.payload?.taskRetryRequestId === claimed.payload.taskRetryIntentId) { // Receipt existence proves the earlier wait, not current authority. // Retire only this unadmitted wake; an admitted run keeps its history. - await db.update(agentWakeupRequests).set({ - status: "failed", error: "chat_failed_run_retry_not_authorized", - finishedAt: new Date(), updatedAt: new Date(), - }).where(and(eq(agentWakeupRequests.id, receipt.id), - eq(agentWakeupRequests.companyId, claimed.companyId), - eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId))); + await db.transaction(async tx => { + await notifyChatInboundWakeNotices(tx, { + companyId: claimed.companyId, issueId: String(claimed.payload.issueId), + }); + await tx.update(agentWakeupRequests).set({ + status: "failed", error: "chat_failed_run_retry_not_authorized", + finishedAt: new Date(), updatedAt: new Date(), + }).where(and(eq(agentWakeupRequests.id, receipt.id), + eq(agentWakeupRequests.companyId, claimed.companyId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), isNull(agentWakeupRequests.runId))); + }); return settle("failed", { code: "failed_chat_retry_not_authorized", attemptCount }); } return settle(["skipped", "failed", "cancelled"].includes(receipt.status) ? "failed" : "processed", { diff --git a/server/src/services/chat-channels/inbound/wakeup-notices.ts b/server/src/services/chat-channels/inbound/wakeup-notices.ts index 3ded0ab4d71..de9f0b57d71 100644 --- a/server/src/services/chat-channels/inbound/wakeup-notices.ts +++ b/server/src/services/chat-channels/inbound/wakeup-notices.ts @@ -18,9 +18,13 @@ type NoticeOptions = { isExternalActionAuthorizationChange: (error: unknown) => boolean; isShuttingDown: () => boolean; }; +export type InboundWakeNoticeScope = { companyId: string; issueId?: string }; +export type InboundWakeNoticeCursor = { createdAt: string; id: string }; +type NoticeCandidate = { action: typeof chatActions.$inferSelect; cursorCreatedAt: string }; + export function createInboundWakeNotices(db: Db, options: NoticeOptions) { const { authorizeInboundWakeup, enqueueFailedChatRetryPublications, authorizeRetainedChatSourcePublication, isExternalActionAuthorizationChange } = options; - let inboundWakeNoticeCursor: { createdAt: Date; id: string } | null = null; + let inboundWakeNoticeCursor: InboundWakeNoticeCursor | null = null; async function currentInboundWakeNotice( tx: DbOrTransaction, action: typeof chatActions.$inferSelect, @@ -130,15 +134,12 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { return Boolean(link); } - async function enqueueInboundWakeupPublications(limit = 50): Promise { - if (options.isShuttingDown()) return 0; - const retryInserted = await enqueueFailedChatRetryPublications(limit); + function noticeCandidateQuery(cursor: InboundWakeNoticeCursor | null, scope?: InboundWakeNoticeScope) { const owner = alias(agentWakeupRequests, "chat_notice_owner"); - const cursor = inboundWakeNoticeCursor; // Keep UUID columns indexable. Match only canonical lowercase UUID text, // preserving the old comparison while malformed payloads match nothing. const payloadUuid = (value: SQL) => - sql`(case when ${value} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then (${value})::uuid end)`; + sql`(case when ${value} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then (${value})::uuid end)`; const payloadIssueId = payloadUuid(sql`${chatActions.payload}->>'issueId'`); const payloadCommentId = payloadUuid(sql`${chatActions.payload}->>'commentId'`); const coalescedOwnerId = sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`; @@ -161,8 +162,9 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`; // A bounded keyset sweep, not an in-memory work queue. Advancing even past // revoked candidates prevents an inaccessible old message starving others. - const candidates = await db - .select({ action: chatActions }) + return db + .select({ action: chatActions, cursorCreatedAt: sql`${chatActions.createdAt}::text`, + issueId: payloadIssueId.as("notice_issue_id") }) .from(chatActions) .innerJoin( agentWakeupRequests, @@ -175,6 +177,9 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { .where( and( eq(chatActions.kind, "inbound_wakeup"), + sql`${payloadIssueId} is not null`, + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(payloadIssueId, scope.issueId) : undefined, inArray(chatActions.status, ["processed", "failed"]), eq(owner.companyId, chatActions.companyId), or( @@ -193,7 +198,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { and ${agentWakeupRequests.status} not in ('cancelled','failed','skipped') and not (${sourceRemoved})) or ${hasVisibleQueue})`, cursor - ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` + ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt}::timestamptz, ${cursor.id}::uuid)` : undefined, sql`not exists (select 1 from chat_publications notice where notice.company_id = ${chatActions.companyId} @@ -203,16 +208,28 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { then 'queued' else 'not_started' end) || ':' || ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`, ), - ) + ); + } + + async function noticeCandidates(limit: number, cursor: InboundWakeNoticeCursor | null, scope?: InboundWakeNoticeScope): Promise { + return noticeCandidateQuery(cursor, scope) .orderBy(asc(chatActions.createdAt), asc(chatActions.id)) .limit(Math.max(1, Math.min(limit, 200))); - const last = candidates.at(-1)?.action; - inboundWakeNoticeCursor = - candidates.length >= Math.max(1, Math.min(limit, 200)) && last - ? { createdAt: last.createdAt, id: last.id } - : null; - let inserted = retryInserted; + } + + function continuation(candidates: NoticeCandidate[], limit: number): InboundWakeNoticeCursor | null { + const last = candidates.at(-1); + return candidates.length === limit && last + ? { createdAt: last.cursorCreatedAt, id: last.action.id } + : null; + } + + async function projectNotices(candidates: NoticeCandidate[], signal?: AbortSignal) { + let inserted = 0; + let contended = false; for (const { action } of candidates) { + signal?.throwIfAborted(); + if (options.isShuttingDown()) throw new Error("Inbound wake notice reconciliation stopped"); try { inserted += await db.transaction(async (tx) => { const [receipt] = await tx @@ -292,15 +309,44 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { return rows.length; }); } catch (error) { - if ( - !isExternalActionAuthorizationChange(error) && - !isExternalChatWaitAuthorizationContention(error) && - !(error instanceof NativeChatReviewPresentationContentionError) - ) - throw error; + if (isExternalChatWaitAuthorizationContention(error) + || error instanceof NativeChatReviewPresentationContentionError) contended = true; + else if (!isExternalActionAuthorizationChange(error)) throw error; } } - return inserted; + return { inserted, contended }; + } + + /** The explicit/manual API keeps its bounded cursor and combined retry-notice + * behavior. Background owners reconcile the two duties independently. */ + async function enqueueInboundWakeupPublications(limit = 50): Promise { + if (options.isShuttingDown()) return 0; + const retryInserted = await enqueueFailedChatRetryPublications(limit); + const pageSize = Math.max(1, Math.min(limit, 200)); + const candidates = await noticeCandidates(pageSize, inboundWakeNoticeCursor); + inboundWakeNoticeCursor = continuation(candidates, pageSize); + return retryInserted + (await projectNotices(candidates)).inserted; + } + + async function loadInboundWakeNoticeScopes(scope?: InboundWakeNoticeScope, signal?: AbortSignal) { + signal?.throwIfAborted(); + if (scope?.issueId) return [{ companyId: scope.companyId, issueId: scope.issueId }]; + const candidates = noticeCandidateQuery(null, scope).as("inbound_notice_candidates"); + return db.selectDistinct({ companyId: candidates.action.companyId, issueId: candidates.issueId }) + .from(candidates); + } + + async function reconcileInboundWakeNotices(input: { + scope: InboundWakeNoticeScope; limit?: number; cursor?: InboundWakeNoticeCursor | null; signal?: AbortSignal; + }) { + // Bound authorization work, not successful publications: denied history is + // still a candidate check. The worker retains this precise cursor between + // passes and resets it when a committed change invalidates the scope. + input.signal?.throwIfAborted(); + const limit = Math.max(1, Math.min(input.limit ?? 100, 200)); + const candidates = await noticeCandidates(limit, input.cursor ?? null, input.scope); + const result = await projectNotices(candidates, input.signal); + return { ...result, nextCursor: continuation(candidates, limit) }; } async function authorizeInboundWakePublication( @@ -380,5 +426,5 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) { throw error; } } - return { enqueueInboundWakeupPublications, authorizeInboundWakePublication }; + return { enqueueInboundWakeupPublications, loadInboundWakeNoticeScopes, reconcileInboundWakeNotices, authorizeInboundWakePublication }; } diff --git a/server/src/services/chat-channels/inbound/wakeups.ts b/server/src/services/chat-channels/inbound/wakeups.ts index 2f7b52d2994..d80fde3d6f9 100644 --- a/server/src/services/chat-channels/inbound/wakeups.ts +++ b/server/src/services/chat-channels/inbound/wakeups.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js"; import { DELIVERY_PROCESSING_STALE_MS } from "./delivery-ledger.js"; import type { EndpointRow, DbOrTransaction } from "../runtime-types.js"; import { notifyChatRunMilestones } from "../../chat-run-milestone-notifications.js"; @@ -113,6 +114,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { await notifyChatDeliveryWork(tx); if (action.status === "processed") { await notifyChatRunMilestones(tx, { companyId: action.companyId, issueId: String(action.payload.issueId) }); + } else if (action.status === "failed") { + await notifyChatInboundWakeNotices(tx, { companyId: action.companyId, issueId: String(action.payload.issueId) }); } const acceptedAt = new Date(); const accepted = await tx @@ -255,6 +258,8 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) { await notifyChatDeliveryWork(tx); if (status === "processed") { await notifyChatRunMilestones(tx, { companyId: claimed.companyId, issueId: request!.issueId }); + } else if (status === "failed" && request) { + await notifyChatInboundWakeNotices(tx, { companyId: claimed.companyId, issueId: request.issueId }); } const changed = await tx .update(chatActions) diff --git a/server/src/services/chat-channels/slack/session-stop.ts b/server/src/services/chat-channels/slack/session-stop.ts index aef5c731362..ea922411c88 100644 --- a/server/src/services/chat-channels/slack/session-stop.ts +++ b/server/src/services/chat-channels/slack/session-stop.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../../chat-inbound-wake-notice-notifications.js"; import type { EndpointRow, DbTransaction, DbOrTransaction, RuntimeContext, LifecycleRuntimeFence } from "../runtime-types.js"; import { dueChatAction, PROVIDER_EFFECT_STALE_MS } from "../../chat-action-work.js"; @@ -584,6 +585,7 @@ export function createSlackSessionStop(db: Db, options: SlackSessionStopOptions) ["queued", "deferred_issue_execution"].includes(wakeup.status) && !issue.executionRunId ) { + await notifyChatInboundWakeNotices(tx, { companyId: current.companyId, issueId: issue.id }); await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/services/chat-channels/task-retry-input.ts b/server/src/services/chat-channels/task-retry-input.ts index 200f77bda1c..71743f42ad4 100644 --- a/server/src/services/chat-channels/task-retry-input.ts +++ b/server/src/services/chat-channels/task-retry-input.ts @@ -2,6 +2,7 @@ import { z } from "zod"; import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm"; import { agentWakeupRequests, chatActions, chatConversations, chatEndpoints, heartbeatRuns, issues, type Db } from "@paperclipai/db"; import type { DbOrTransaction } from "./runtime-types.js"; +import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; export const taskRetryInputSchema = z.object({ // This is a server-selected snapshot of the durable queue, not client input. @@ -89,6 +90,7 @@ export async function releaseUnadmittedTaskRetryInput(db: Db, input: { companyId eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.id, input.retryActionId), )).for("update"); if (receipt?.runId) return; + await notifyChatInboundWakeNotices(tx, { companyId: input.companyId, issueId: issue.id }); await tx.update(agentWakeupRequests).set({ status: "deferred_issue_execution", finishedAt: null, payload: sql`${agentWakeupRequests.payload} - 'coalescedIntoWakeupRequestId'`, updatedAt: new Date(), }).where(and(eq(agentWakeupRequests.companyId, input.companyId), eq(agentWakeupRequests.status, "coalesced"), diff --git a/server/src/services/chat-inbound-wake-notice-notifications.ts b/server/src/services/chat-inbound-wake-notice-notifications.ts new file mode 100644 index 00000000000..12e0aa08422 --- /dev/null +++ b/server/src/services/chat-inbound-wake-notice-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import type { InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js"; + +export function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +} diff --git a/server/src/services/chat-inbound-wake-notice-work.ts b/server/src/services/chat-inbound-wake-notice-work.ts new file mode 100644 index 00000000000..3181e409cd2 --- /dev/null +++ b/server/src/services/chat-inbound-wake-notice-work.ts @@ -0,0 +1,44 @@ +import type { ChatChannelService } from "./chat-channels.js"; +import type { InboundWakeNoticeCursor, InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; + +type Entry = { + id: string; companyId: string; issueId: string; dueAt: number; + cursor: InboundWakeNoticeCursor | null; contended: boolean; +}; +const batchSize = 100; + +/** Admission receipts own queued/not-started/removed notices. The channel + * service retains authorization and publication policy; this owner schedules + * only unfinished batches, contention, and failed reconciliation. */ +export function chatInboundWakeNoticeWork(service: Pick) { + async function load(scope?: InboundWakeNoticeScope, signal?: AbortSignal): Promise { + return (await service.loadInboundWakeNoticeScopes(scope, signal)).map(row => ({ + ...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0, cursor: null, contended: false, + })); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + const result = await service.reconcileInboundWakeNotices({ + scope: { companyId: entry.companyId, issueId: entry.issueId }, + limit: batchSize, cursor: entry.cursor, signal, + }); + const contended = entry.contended || result.contended; + if (result.nextCursor) return { ...entry, cursor: result.nextCursor, contended, dueAt: Date.now() }; + // Finish the pass before revisiting contention, so a locked prefix cannot + // starve later candidates. Do not forget an earlier page's unresolved lock. + return contended ? { ...entry, cursor: null, contended: false, dueAt: Date.now() + 1000 } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatInboundWakeNotices, + decodeScope: (hint: string): InboundWakeNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: InboundWakeNoticeScope) => !scope || + (entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId)), + load, check, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index d3f38e52d81..6964bd6b069 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +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"; @@ -10,4 +11,6 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston // Lifecycle and coarse native progress share lineage and authority. Extend // this eligibility boundary once rather than adding wrappers to its callers. await notifyChatNativeProgress(tx, scope); + // Admission and current chat authority also govern saved queue notices. + await notifyChatInboundWakeNotices(tx, scope); } diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 580df76b6e9..1534817fa2e 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { isRunnerApiResponseAsset } from "./asset-content-access.js"; @@ -65,6 +66,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), )); + await notifyChatInboundWakeNotices(tx, { companyId: id }); await tx .update(agentWakeupRequests) .set({ @@ -332,6 +334,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (companyPatch.status !== undefined && companyPatch.status !== existing.status) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyCompanyRetentionWork(tx, id); + await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); } const updated = await tx @@ -484,6 +487,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) { if (!wasAlreadyArchived) { await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyCompanyRetentionWork(tx, id); + await notifyChatInboundWakeNotices(tx, { companyId: id }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id); await tx .update(companies) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 45f9e520f8d..d652eef12f2 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -58,6 +58,7 @@ export const DELIVERY_QUEUES = { chatInbound: "chat-inbound", chatPublications: "chat-publications", chatNativeProgress: "chat-native-progress", + chatInboundWakeNotices: "chat-inbound-wake-notices", chatRunMilestones: "chat-run-milestones", chatReceipts: "chat-slack-receipts", browser: "browser-use", diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index c32cebc2013..a6d57b7d46a 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { resumeDeferredChatWake } from "../durable-chat-wakeup.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { readNativeStopExecutionEvidence } from "./native-stop-evidence-read.js"; @@ -1016,6 +1017,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont if (wakeupIds.length === 0) return 0; + await notifyChatInboundWakeNotices(tx, { companyId: scope.companyId }); await tx .update(agentWakeupRequests) .set({ @@ -1576,35 +1578,29 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont ); if (uniqueAgentIds.length === 0) return 0; - const now = new Date(); - const wakeupIds = await db - .select({ id: agentWakeupRequests.id }) - .from(agentWakeupRequests) - .where( - and( + return db.transaction(async tx => { + const wakeups = await tx + .select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId }) + .from(agentWakeupRequests) + .where(and( inArray(agentWakeupRequests.agentId, uniqueAgentIds), - inArray(agentWakeupRequests.status, [ - "queued", - "deferred_issue_execution", - ]), - sql`${agentWakeupRequests.runId} is null`, - ), - ) - .then((rows) => rows.map((row) => row.id)); - - if (wakeupIds.length === 0) return 0; - - await db - .update(agentWakeupRequests) - .set({ - status: "cancelled", - finishedAt: now, - error: reason, - updatedAt: now, - }) - .where(inArray(agentWakeupRequests.id, wakeupIds)); - - return wakeupIds.length; + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), + )); + if (!wakeups.length) return 0; + for (const companyId of new Set(wakeups.map(row => row.companyId))) { + await notifyChatInboundWakeNotices(tx, { companyId }); + } + const now = new Date(); + const cancelled = await tx.update(agentWakeupRequests).set({ + status: "cancelled", finishedAt: now, error: reason, updatedAt: now, + }).where(and( + inArray(agentWakeupRequests.id, wakeups.map(row => row.id)), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), + )).returning({ id: agentWakeupRequests.id }); + return cancelled.length; + }); } async function cancelInvocationsForAgentsInternal( diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 1d8af92fd5b..227fae5948f 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -1198,28 +1199,31 @@ export function issueTreeControlService(db: Db) { const issueIds = treeIssues.map((issue) => issue.id); if (issueIds.length === 0) return []; const now = new Date(); - return db - .update(agentWakeupRequests) - .set({ - status: "cancelled", - finishedAt: now, - error: reason, - updatedAt: now, - }) - .where( - and( - eq(agentWakeupRequests.companyId, companyId), - inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), - isNull(agentWakeupRequests.runId), - inArray(sql`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds), - ), - ) - .returning({ - id: agentWakeupRequests.id, - agentId: agentWakeupRequests.agentId, - reason: agentWakeupRequests.reason, - payload: agentWakeupRequests.payload, - }); + return db.transaction(async tx => { + await notifyChatInboundWakeNotices(tx, { companyId }); + return tx + .update(agentWakeupRequests) + .set({ + status: "cancelled", + finishedAt: now, + error: reason, + updatedAt: now, + }) + .where( + and( + eq(agentWakeupRequests.companyId, companyId), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]), + isNull(agentWakeupRequests.runId), + inArray(sql`${agentWakeupRequests.payload} ->> 'issueId'`, issueIds), + ), + ) + .returning({ + id: agentWakeupRequests.id, + agentId: agentWakeupRequests.agentId, + reason: agentWakeupRequests.reason, + payload: agentWakeupRequests.payload, + }); + }); } return { diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index aa550c775a3..8185a4881c5 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; import { notifyDecisionTargetWork, notifyDecisionTargetIssueRemoval } from "./decision-targets.js"; @@ -11316,6 +11317,9 @@ export function issueService(db: Db) { }); } if (changesPrivacy) { + // Privacy inherited through task ancestry can authorize saved notices + // on descendants as well as this task. + await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId }); const nextProjectId = issueData.projectId !== undefined ? issueData.projectId : receiptExisting.projectId; const [privacyProject] = nextProjectId ? await tx.select().from(projects) .where(and(eq(projects.id, nextProjectId), eq(projects.companyId, existing.companyId))) : []; @@ -11397,6 +11401,7 @@ export function issueService(db: Db) { // Invalidate human direction at the common assignment boundary, including // plugin/service writes that do not go through HTTP run cancellation. // Keep the requester attribution for audit; cancellation revokes its use. + await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId }); await tx.update(agentWakeupRequests).set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() }) .where(and(eq(agentWakeupRequests.companyId, receiptExisting.companyId), eq(agentWakeupRequests.requestedByActorType, "user"), @@ -11425,6 +11430,9 @@ export function issueService(db: Db) { if (patch.status === "done" || patch.status === "cancelled") { await notifyDecisionTargetWork(tx, `issue:${id}`); } + if (patch.status !== undefined && patch.status !== receiptExisting.status) { + await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId, issueId: id }); + } if (patch.hiddenAt === null && receiptExisting.hiddenAt !== null) { await notifyChatRunMilestones(tx, { companyId: receiptExisting.companyId, issueId: id }); } @@ -12615,6 +12623,11 @@ export function issueService(db: Db) { return db.transaction(async (tx) => { if (!(await lockIssueForCommentMutation(tx, commentId))) return null; + const [source] = await tx.select({ companyId: issueComments.companyId, issueId: issueComments.issueId }) + .from(issueComments).where(and(eq(issueComments.id, commentId), isNull(issueComments.deletedAt))) + .for("update"); + if (!source) return null; + await notifyChatInboundWakeNotices(tx, source); const now = new Date(); const [comment] = await tx .update(issueComments) diff --git a/server/src/services/projects.ts b/server/src/services/projects.ts index d70b26626d0..28cae661e9b 100644 --- a/server/src/services/projects.ts +++ b/server/src/services/projects.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { listProjectWorkspaceProviders, resolveProjectWorkspaceProvider } from "./project-workspace-provider.js"; import { retireRemovedPluginWorkspaceDependents } from "./plugin-workspaces.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -80,12 +81,11 @@ export async function ensureProjectAccessMember( subjectId: string; }, ) { - return dbOrTx - .insert(projectAccessMembers) - .values(input) - .onConflictDoNothing() - .returning() - .then((rows: Array) => rows[0] ?? null); + return dbOrTx.transaction(async (tx: Db) => { + await notifyChatInboundWakeNotices(tx, { companyId: input.companyId }); + return tx.insert(projectAccessMembers).values(input).onConflictDoNothing().returning() + .then(rows => rows[0] ?? null); + }); } /** @@ -1013,6 +1013,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool const row = await db.transaction(async (tx) => { if (updates.visibility !== undefined) { + await notifyChatInboundWakeNotices(tx, { companyId: existingProject.companyId }); await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`issue-privacy-tree:${existingProject.companyId}`}, 0))`); } const [updated] = await tx.update(projects).set(updates).where(eq(projects.id, id)).returning();