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();