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 f9bebd56ecd..85c7f468f0a 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2204,3 +2204,49 @@ wake notice audit for the producer inventory, regressions and fixture correction 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. + + +## Slice 30 gate and slice 31 projection checkpoint (2026-10-25) + +[#17647](https://github.com/paperclipai/paperclip/pull/17647) at +`fee07f3fa8cbc489cb9bbf657a6065c6e00748c3` has all 53 checks passing, +a fresh Greptile 5/5 review, and its fairness finding resolved. The pre-repair +full chat/startup acceptance passed 1,486 cases; the bounded-candidate repair +passes 44 selected integration cases and server typecheck/build. No merge occurred. + +The next separate branch, `codex/sleep-31-failed-retry-notices`, fixes a reproduced +manual-cursor starvation defect and adds scoped, bounded retry-notice projection. +Its normal periodic pass remains until producer/lifecycle migration is complete. +See [the retry notice audit](2026-10-25-failed-retry-notices-sleep-audit.md). + +Fresh older-parent audit still shows #17490 at `31bcd19888` and #17572 at +`8815dfad38` failing E2E shard 3 and its aggregate. #17576 still has unresolved +voice inventory amplification findings 4371283330/4371307607, and #17580 still +has unresolved company-offboarding/cleanup findings 4370459800/4370543984. These +are required repairs, not exemptions. The full original migration scope, +master reconciliation, and all latest-head gates remain required. + + +## Slice 31 worker and producer checkpoint (2026-10-25) + +Failed-retry notices now register a scoped worker and leave periodic publication +maintenance. Real missed wakes were reproduced for admission, agent/tree +cancellation, resumed-admission denial, early admission refusal, and requester +restoration. Their owning transactions now notify through the existing shared +boundary. Fourteen selected lifecycle cases and all 1,503 chat/startup acceptance +cases pass. Workspace typecheck and build pass. See the retry audit for retained +failures, exact commands/logs, and native source-cleanup qualification still open. +This is a draft checkpoint, not a merge-ready claim. Older parent gates and every +remaining original recurring-server duty stay in scope. No merge occurred. + + +## Slice 31 native producer repairs + +Draft [#17649](https://github.com/paperclipai/paperclip/pull/17649) now covers +source lease cleanup and reusable-lease retirement, each with a reproduced +missing notice and real writer transaction tests. All 36 selected native/notice +cases pass. The owning audit distinguishes notice authority from the stronger +filesystem execution proof, and retains the fake-clock connection failure and +Runner CI receipt timeout. Existing lease acceptance, updated full acceptance, +remote gates, older parent findings, and every remaining original duty are still +required. No stack PR has been merged. diff --git a/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md b/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md new file mode 100644 index 00000000000..c7b92fcb356 --- /dev/null +++ b/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md @@ -0,0 +1,450 @@ +# Failed-run retry notice migration + +Date: 2026-10-25. Branch: `codex/sleep-31-failed-retry-notices`. +Parent: [#17647](https://github.com/paperclipai/paperclip/pull/17647), +`fee07f3fa8cbc489cb9bbf657a6065c6e00748c3`. +This is slice 31 of the recurring-server-work stack. The branch now registers +a scoped worker and removes retry notices from periodic publication maintenance. +Producer audit and full acceptance remain in progress; this is not merge-ready. +Draft PR: [#17649](https://github.com/paperclipai/paperclip/pull/17649). + +## Obligation and boundary + +A failed-run retry may have a durable wake receipt but no admitted run. Its +existing `chat_actions` row has kind `failed_run_retry`; the same-ID, +same-company `agent_wakeup_requests` row is authoritative for whether the retry +waits or was declined. The notice is eligible while `runId` is null and the +receipt is deferred, cancelled, failed, or skipped. A matching existing +publication discharges that state. Queued and not-started are separate saved +states; a later cancellation can owe a second notice. + +The retry service owns current source/batch authorization, task Retry generation, +original principal, retained source digest, endpoint reach, provider binding, +execution containment, and retry admission. The new projection module receives +a narrow authorization callback returning the destination, task and source +comment IDs. It does not reproduce that policy or add provider branches. +Existing publication transport still authorizes immediately before I/O. + +The new projection module is `chat-channels/failed-retry-notices.ts`. It owns +candidate selection, receipt locking, deduplicated insertion, exact cursor +precision and scoped reconciliation. The service retains its existing public +combined manual notice API. Nothing adds a database table or a framework adapter. + +## Current implementation and evidence + +- The manual cursor previously converted `createdAt` to JavaScript Date. A full + denied page with PostgreSQL microseconds was selected repeatedly, starving + a later valid retry. The regression reproduces through the public combined + API in `/tmp/sleep31-cursor-before-public-api.log`: the second call inserts + zero instead of the expected one notice. +- `/tmp/sleep31-cursor-before.log` is an invalid first reproduction that attempted + to call a private service method. It is retained as failure evidence, not + evidence of the cursor defect. +- Cursors now use PostgreSQL `timestamp::text` plus the UUID tiebreaker. Manual + selection retains its existing maximum batch of 50; scoped reconciliation + bounds candidate authorization work, not successful publications. +- Permanent authority rejection settles the pass. Actual PostgreSQL/authority + contention is returned separately for the future worker to retain. Unknown + errors propagate. Run-bound and already projected states leave no candidates. +- Five integration cases pass in `/tmp/sleep31-scoped-notices.log`: manual/scoped + denied-prefix pagination, scoped isolation and receipt contention/cancellation, + and the existing Slack/Telegram queued/working/selected-final identities. + The scoped cursor test observes exactly 50 then one candidate transaction. +- Server typecheck passes in `/tmp/sleep31-scoped-typecheck.log`. + This establishes the projection contract, not idle worker behavior or host + suspension. Full acceptance, worker registration, CI and review remain open. + +## Producer audit to complete + +- [ ] Receipt admission/defer: trace actual heartbeat and wake-queue writers, + including the ordinary task Retry intent and the chat action receipt. A + callback returning a truthy result is not sufficient persistence evidence. +- [ ] Receipt cancellation/failure: task tree, agent, budget, company, Slack Stop, + deferred queue writers, and retry-specific authorization denial. Reuse the + existing transaction signals where possible; test outer rollback and commit. +- [ ] Admission/promotion and settlement: run-bound receipts stop owing notices; + retain notification ordering through transaction commit and failure. +- [ ] Retry action lifecycle: staged, claimed, processed, failed, cancelled, + recovered, and superseded generations; determine which transitions can make + current source authorization succeed or fail. +- [ ] Current authority restoration: endpoint/provider runtime, resources, + linked identity, membership, task/project privacy and explicit grants, and + original source/run lineage. Use existing eligibility signals rather than + scattering new wrappers through these owners. +- [ ] Source and generation changes: inspect task Retry intent rewrites, + recovery completion, preserved source metadata, and repaired delivery records. +- [ ] Mixed ordinary/retry receipt ownership: `findExistingDeferredWake` does + not exclude a failed-retry owner by kind. A retry's own `allowRunCoalescing: + false` does not establish what a later ordinary wake can select. The ordinary + retirement path requires an exact visible queued publication; retry-owned + publication authorization additionally requires its comment ID to equal the + final original retry source comment. Establish the reachable combinations with + real admission before exempting the retry denial writer from ordinary signals. +- [ ] Independent clients/replicas: current process-local signals cover only the + supported owning application client and child pools. Do not claim external + SQL or another replica is observed. + +## Worker and acceptance still required + +- [ ] Register a scoped deadline worker using the shared coordinator, retaining + exact page cursors and contention across pages. Committed invalidation must + restart a cursor so restored earlier candidates are revisited. +- [ ] Empty/configured-idle/retained-history quiet, real first admission, actual + producer cancellation, denied history, lock contention, retry after failure, + rollback/outer commit, restart, drain/resume, shutdown, and writes arriving + during selection and pagination. +- [ ] Preserve all existing provider identity and authorization acceptance. +- [ ] Remove only retry notices from periodic maintenance after the above passes. +- [ ] Full required verification, template PR, latest-head checks and fresh 5/5 + review with no unresolved findings. Do not merge the stack. + +No change to Cloud sleep-safety gates or external wake coverage is proposed. +Terminal confirmations, interaction continuations and GitHub maintenance remain +separate duties. The rest of the original migration ledger remains in scope. + + +## Broader acceptance and retained failure evidence + +Server build passed in `/tmp/sleep31-scoped-build.log`. Broader selection +`retry|Retry|pre-run FIFO` first finished with 180 passing and one failing case +in `/tmp/sleep31-retry-acceptance.log`. The failing test injects the first +transactional agent-status read error while publishing a selected final. It +instead reached retry-notice maintenance because the reused test database still +contained two deferred retry candidates from the deliberate failing reproductions. +Direct SQL identified those candidates at 13:33:57 and 13:34:35, matching the +invalid first reproduction and the valid before-fix cursor reproduction. Neither +had a publication. This is fixture contamination, not an inferred flake from a +later success. + +The cursor fixture now removes its own original staged action/receipt as well +as its denied clones in `finally`, including on a failed assertion. The original +database is retained as evidence. After applying migrations to a fresh isolated +`paperclip_sleep31_retry_acceptance_20261025_test`, all 181 selected cases pass +in `/tmp/sleep31-retry-acceptance-fresh.log`. No production error handling was +relaxed to bypass the failure. The local timestamp/fixture evidence is test-only. + +Additional producer audit findings to resolve before wiring: + +- The admission reader can select a previously deferred retry as a later + ordinary wake's owner; the current ordinary publication path requires a + matching visible original queue notice, while retry-owned transport requires + its original final source comment. Reachability must be established, not + assumed from the retry's own no-coalescing option. +- Wake-queue `cancelDeferredWake` and `failDeferredWake` signal the ordinary + inbound-notice helper rather than the broader milestone helper. A retry + notification added only to milestones would miss those real writers. +- `normalizeDeferredWakeCommentIds` changes the saved payload without a notice + signal. Check whether removing merged/deleted input restores a retry's exact + source eligibility before classifying it as harmless bookkeeping. +- `retireSupersededTaskRetry`, `requestTaskRetry`, and retry reconciliation write + the separate task-intent receipt. The notice candidate uses the chat-action + receipt; examine generation/source authorization before adding blanket wakes + to all intent timestamp writes. + + +## Admission and cancellation migration checkpoint + +The worker now uses the shared scoped-deadline coordinator. It reconstructs +existing receipt obligations at startup, resets an issue cursor on a committed +invalidation, and retains bounded pages, real contention, and failed selection. +Settled and denied history has no recurring scan. No new table is added. + +`notifyChatInboundWakeNotices` now also notifies the retry-notice topic. This +keeps task-tree, agent, budget, company, deferred-queue, source/privacy, and +existing milestone authority signals at one boundary. The hint serializes only +company and optional issue IDs. The new worker is absent from older parents; +there it creates no new timer or database listener. + +New producer changes are narrow: + +| Owner | Obligation and notification | +| --- | --- | +| Retry admission callback | After current authorization succeeds, register intent inside the heartbeat transaction before the receipt write. | +| Retry authorization denial | Retire the exact unadmitted receipt in a transaction that registers notice intent before its conditional update. No provider I/O enters the transaction. | +| Heartbeat early refusal | A durable chat receipt can be skipped before its authorization callback. Signal its existing execution-wait transaction, or create one transaction for the previously standalone skipped-row insert. Non-chat callers retain their existing path. | +| Deferred-queue requester merge | A manual requester replacement can restore or revoke the original receipt identity. Register intent before that update. Payload-only merges do not add a notification. | + +Causal evidence retained: + +- `/tmp/sleep31-cancellation-before.log`: agent cancellation missed the notice; + the tree case initially used an invalid direct function reference. +- `/tmp/sleep31-tree-cancellation-before.log`: corrected real task-tree service + also reproduces the missing notice. The shared notification repair passes + 39 related cases in `/tmp/sleep31-cancellation-fixed.log`. +- `/tmp/sleep31-real-admission-lifecycle.log`: real heartbeat admission missed + the notice in normal, drain, and failed-selection modes. Restart alone found + it. The callback notification fixes that causal gap. +- `/tmp/sleep31-real-admission-fixed.log`: one normal-path assertion observed + an unexpected worker error without printing it. This error was not classified + as a flake. The assertion now shows full errors. The diagnostic run instead + exposed the expected failure's AggregateError wrapper; that expectation now + checks its nested cause. Retain the first observation for broad acceptance. +- `/tmp/sleep31-denial-before.log`: the actual recovery service restages a + deferred retry; a controlled admission rejection saves `failed`, but no + not-started notice appears. The conditional transaction notification fixes it. +- `/tmp/sleep31-early-refusal-before.log`: real heartbeat admission with + wake-on-demand disabled saves a skipped receipt but misses its notice. The + early-refusal writer notification fixes it. +- `/tmp/sleep31-requester-before.log`: the real deferred-queue Postgres writer + restores the original requester but misses the notice. This exercises the + producer transaction, not a synthetic notification. It does not establish + every end-to-end ordinary/retry coalescing path. +- `/tmp/sleep31-producer-lifecycle.log`: all 14 selected cases pass, including + real admission, first-write drain/resume, failed selection followed by drain, + stop/restart, agent/tree rollback and commit, admission denial, early refusal, + privacy restoration after idle and during a bounded page, and requester merge. + The quiet assertions observe no root SQL/transaction calls for 120 seconds. +- `/tmp/sleep31-workspace-typecheck.log` and + `/tmp/sleep31-workspace-build.log`: workspace typecheck and build pass. + +The full chat/startup suite passes all 1,503 cases on fresh migrated +`paperclip_sleep31_wired_chat_20261025_test`, with evidence in +`/tmp/sleep31-wired-full-chat.log` (456.47 seconds, exit 0). This includes +the existing provider authorization and publication acceptance suite. +The stopped-worker case also checks that publication maintenance no longer +projects retry notices before startup reconstruction resumes the worker. + +### Remaining producer qualifications + +`readTaskChatRetryIntent` uses saved identity, generation, source, actor and +binding. It does not use the separate task-intent status or updatedAt. Ordinary +status reconciliation therefore does not itself restore notice eligibility. +Restaging the chat action from processed to issued also stays within the set of +statuses already allowed for publication. Changes that bind a receipt to a run +remove it from this notice obligation; the projection rechecks before writing. + +`assertFailedRetryReceipt` validates stable receipt identity, not normalized +queued-comment payload. Payload-only comment normalization is not an authority +restoration for retry notices. The requester replacement is different and is +now covered at its actual writer. Existing ordinary-source notice behavior is +still included through the shared signal fanout. + +Native exact-retry source authorization needs further qualification before +readiness: it reads native finalization ownership, cleanup history, environment +lease settlement, and retained checkpoint proof. `native-session-executor.ts` +commits a settled maintenance history entry and then removes its activation +file; `environments.ts::releaseLease` uses environment-cleanup notifications. +Determine which of these can restore an already-existing retry notice and prove +the required wake after the full authority boundary. A signal before the final +filesystem handoff may be too early. Do not claim all native cleanup producers +are covered by the general milestone fanout merely because run finalizers use it. + +All older stack failures and original recurring-server duties remain required. +No merge, expanded Cloud wake coverage, or deployed host sleep is claimed. + + +## Native lease producer qualification + +Two real producer gaps are now repaired without changing retry authority: + +- `environmentService.releaseLease` can settle a source lease after a notice + pass was denied. It now locks the exact matching row and registers the + existing scoped notice intent before changing eligibility fields. The + pending-cleanup attempt predicate remains on both selection and update; + stale claims cannot release the resource or create a notice wake. Unchanged + release repeats do not notify this worker. The existing environment-cleanup + transaction remains the owner, including caller-owned outer transactions. +- `environmentService.acquireLease` can retire a retained predecessor when + transferring a reusable sandbox to a new run. It registers a company-scoped + intent in that existing transaction before retirement. The predecessor can + belong to another issue in the same execution workspace, so the new issue is + not a sufficient scope. The original company/environment/resource/ownership + predicates and rollback on failed handoff remain unchanged. The notice worker + still bounds candidate authorization per pass. + +Both regressions start with an admitted retry receipt, make its native source +ineligible through a lease, settle the denied worker, then use the actual service +writer. They cover stale claims, outer rollback and commit, repeat writes, one +publication, and 120 seconds of logical idle with no root SQL or transactions. +The retained before-fix logs are `/tmp/sleep31-native-cleanup-before.log` and +`/tmp/sleep31-replacement-before.log`. Atomic release passed 35 selected cases +in `/tmp/sleep31-native-cleanup-atomic.log`. Both producers and the other notice +lifecycle cases pass 36 tests in `/tmp/sleep31-native-lease-producers-fixed.log`. + +The intermediate combined run `/tmp/sleep31-native-lease-producers.log` passed +the new producer cases but failed the disabled-admission case with a recorded +PostgreSQL `CONNECT_TIMEOUT`. Its fake clock advanced 1,000ms on each wait while +real TCP connection work was pending. Those waits now advance 10ms; the same +strict error assertions and 120,000ms settled-quiet assertion pass. This supplies +a concrete cause for this failure, not proof that the earlier unprinted error +had the same cause. Keep that earlier observation in the acceptance record. + +### Native cleanup boundary disposition + +The notice projection calls `authorizeFailedChatRunRetry(publication: true)`. +For a non-task Retry this reaches `failedNativeRetryCoordinator`, which reads +saved native ownership, recovery and environment lease state. It does not call +`assertFailedNativeRetryState` or its filesystem activation/checkpoint verifier. +The stronger filesystem check remains on initial preparation and admission. +Task Retry uses its saved intent authority and intentionally bypasses the +non-task coordinator check for status publications; it does not gain execution +permission from the notice worker. Therefore filesystem activation completion +is not itself a new notice-eligibility producer. + +For the retired-owner recovery path, initial non-task retry preparation already +requires settled native maintenance and the physical checkpoint proof. The +maintenance service refuses another ordinary pass over a nonempty canonical +root with existing maintenance history; its interrupted-copy path requires +exactly the earlier incomplete history and cannot recreate a settled generation. +The first successful cleanup precedes creation of that non-task retry action. +Do not add a pre-activation notification to filesystem cleanup on the assumption +that all execution predicates are also notice predicates. Existing source +finalization writers use the milestone notification boundary; source lease +release and reuse were the actual missing producers found here. + +`releaseLeasesForRun` has no production caller in the inspected repository; +only its definition and environment-service test call it. `extendLease` changes +expiry/last-used metadata, which these notice authorization predicates do not +read. Same-run reacquisition makes a lease active (revokes eligibility); its +later settlement uses the covered release writer. Current publication authority +is still rechecked before sending. Future producers must use the same committed +notification contract rather than relying on this audit as a blanket exemption. + +### Remote and local gate record + +At `cbfaab918a`, Greptile reports 5/5 with no inline findings, explicitly leaving +the native audit unresolved. CI run 40224785432 has a Runner failure in +`local-runner.test.ts`: duplicate command receipt `command_happy_path_3` timed +out. Evidence: `/tmp/sleep31-ci-runner-failure.log`, job 121263785884. +The first local attempt was invalid because debug Rust binaries were missing +(`/tmp/sleep31-runner-receipt-check.log`). After the prescribed `build:rust`, +all 10 cases pass in `/tmp/sleep31-runner-receipt-built.log`. This does not alone +clear remote CI. No unchanged-job retry has been used for this PR. + +Server typecheck passed with the two lease producers in +`/tmp/sleep31-native-lease-types.log`. The environment-service suite now accepts +an explicitly named isolated external test database ending in `_test`, retaining +embedded startup by default and closing the external client afterward. This +allows verification of the existing lease contracts on this host. Its result, +full updated acceptance, latest-head review and CI still need recording. + + +Existing environment-service acceptance passes all 51 cases on fresh migrated +`paperclip_sleep31_environment_20261025_test` in +`/tmp/sleep31-environment-service.log`. Server typecheck and build pass at this +lease-repair checkpoint (`/tmp/sleep31-native-final-types.log` and +`/tmp/sleep31-native-final-build.log`). Updated full chat acceptance and remote +checks remain required. Initial remote head `cbfaab918a` finished with only the +Runner receipt timeout and dependent `ci / verify` failing; all other 51 checks +passed. The forthcoming source-change push must earn its own checks and review. + + +## Master integration: budget-refusal fixture transaction + +The combined full chat suite passed 1,761 cases and failed the newer-saved-input +case that remains budget-blocked. A focused run reproduced that failure among +49 related cases. Its `failed_chat_retry_contended` receipt was not evidence of +actual database contention: temporary diagnostic instrumentation captured +`Database work must be registered inside a transaction` from the budget mock's +`request.authorize(db)` call. The real admission owner passes a transaction. + +The mock now authorizes and writes its skipped wake receipt in one transaction. +All original requester, source-comment, and skipped-status assertions remain. +No production retry, budget, or authority behavior changed. All 49 related cases +pass on this owner. The initial full-suite and focused failures remain recorded +in `/tmp/sleep-master-chat-full-tests.log` and `/tmp/sleep-master-chat-budget-tests.log`; +the captured cause and corrected result are in +`/tmp/sleep-master-budget-fixture-diagnostic.log` and +`/tmp/sleep-master-budget-fixture-fixed.log`. Temporary diagnostic logging was +removed before this commit. The full combined suite still requires a fresh run. + + +## Agent lifecycle reactivation + +A parked failed-retry notice also becomes eligible when its agent finishes +resuming. The lifecycle transaction now uses the shared `beforeStatusChange` +effect before changing a non-invokable status to an invokable status. The service +supplies the existing company-scoped inbound/failed-notice notification; the +lifecycle module owns status transitions and knows no chat/provider policy. This +brings forward the same lifecycle effect used by downstream native response work, +without adding a polling fallback, table, or caller recovery wrapper. + +The real retry-notice integration first failed because a successfully resumed +agent left the parked notice unpublished (`/tmp/sleep-agent-resume-red.log`). All +four authority-reactivation cases now pass, including two minutes of virtual +quiet before and after the real pause/resume producer +(`/tmp/sleep-agent-resume-green.log`). The full lifecycle suite passes all 64 +cases, including outer transaction commit/rollback and no notification on an +unchanged ready state (`/tmp/sleep-agent-resume-lifecycle-fixed.log`). The first +lifecycle regression fixture incorrectly hired an initially paused agent, whose +resume target is preparation; it was corrected to hire, complete preparation, +pause, and resume through the real lifecycle. Its initial failure is retained in +`/tmp/sleep-agent-resume-lifecycle.log`. Server typecheck passed in +`/tmp/sleep-agent-resume-types.log`. These results establish worker and transaction +behavior only; final-stack acceptance and latest-head CI still apply. + + +## Current integration and discovery acceptance — October 28 + +The prepared local slice follows current inbound parent +`661f2f6da8c6c22cdcc25edccd57502ee6fa8c4b`, including actual master +`a2fbce9305c48f5242e6bafb19d049d29edc1d72`. Normal merges preserve the +19-path slice. Two conflicts retained current lifecycle fixture migrations/client +cleanup and the extracted retry notice import; duplicate imports were removed. +The identical producer helper copied forward to the parent is retained once at +this slice's original location, with both parent producer regressions intact. +An initial raw binary-patch comparison failed after parent test additions changed +blob hashes and hunk offsets. A full-context comparison excluding only those +representation fields passes; `/tmp/sleep31-discovery-parent-update-proof.json` +records the exception rather than claiming raw binary identity. + +The discovery-cost finding in #17647 also existed in this owner's page loop. +Three new real PostgreSQL regressions with 250 retained candidates and a second +company fail before the fix. Known task hints now require zero discovery SQL; +company/startup discovery selects DISTINCT company/task identifiers using the +same candidate predicate as bounded projection. The first SQL version failed +with PostgreSQL 42702 because both joined tables project company_id; an explicit +projection alias fixes the ambiguity. The final four targeted discovery/history +cases pass. This does not claim constant database work for broad discovery. + +Current focused acceptance passes **404 distinct cases** on the integrated owner: +259 across six complete consumer/lifecycle/environment suites, 20 failed-retry +notice and native-lease producer cases, 53 ordinary-notice compatibility cases, +and 72 separate retry/budget compatibility cases. The three chat selections skip +1,651, 1,618 and 1,599 cases respectively; skipped cases are not passes. +Each group uses separate PostgreSQL fixtures; lifecycle and control apply their +own migrations, while chat and environment fixtures are explicitly premigrated. +Final server typechecking/build and module boundaries pass. + +Exact commands/fixtures are `/tmp/sleep31-final-acceptance-command.json`; logs and +exit receipts are `/tmp/sleep31-final-{consumers,notices,ordinary,compatibility}.*` +and `/tmp/sleep31-alias-final-{types,build}.*`. The original query-budget failures, +first ambiguous-column failure, and final green result remain in +`/tmp/sleep31-discovery-{red,green,alias-green}.log`. Prior historical counts and +failures above retain their original revision scope. This local preparation has +not been published, and the parent has a new current-head P1 about ordinary input +restoration that must be repaired and adopted before this slice can merge. +Fresh exact-head remote CI/review, full repository acceptance and deployed sleep +are separate claims; no live provider or physical sleep result is asserted here. + + +## Final landed-parent acceptance — October 28 + +The preceding inbound notice PR #17647 has now merged at +`1aa3a21ac41f602977638cf0be875b01a243e69f`, including both discovery and restored-input +wake repairs. The current normal two-parent integration is +`43d0f84ea56845b8fc4946a7533ecbf7f07b679a`, with parents the prepared retry slice +`c56cc27611c76587a278ac175470e0b4646b7b5f` and actual master. Squash ancestry makes +the ordinary historical merge report eight conflicts. Using the verified landed +parent `41f7e2c9e351efa6cfa5ef699c0e69df7425012b` as the semantic merge base yields +a clean tree. The full 19-path owned binary patch is identical and all 23 unrelated +master paths/deletions are preserved. That proved tree resolves the ancestry-only +conflicts in an ordinary two-parent merge; no rebase or force push is used. + +Current combined-master acceptance passes **42 distinct cases**: 21 retry-notice, +native-lease and restored-input producer cases (1,651 skipped), plus all 21 cases +in app/startup and task-drain admission release suites. The newly landed sandbox +resume/admission behavior is exercised with this slice's queue notifications. +Checks run after server typechecking/build and module-boundary verification to +avoid competing cold-import setup loads. All those checks pass. Commands/fixtures: +`/tmp/sleep31-landed-acceptance-command.json`; results/receipts: +`/tmp/sleep31-landed-{types,build,boundaries,notices,consumers}.*`. + +The earlier 404-case matrix and subsequent 21-case parent check retain their +original revision scopes and are not counted as freshly rerun. All owned source +and test files remain unchanged through the current normal integration; only +this acceptance documentation follows the current tests. Their 17 hashes are in +`/tmp/sleep31-landed-tested-source-sha256.json`; the complete integration proof is +`/tmp/sleep31-landed-parent-publication-proof.json`. Fresh exact-head remote checks +and Greptile 5/5 remain required after publishing this updated branch. There is +no full local repository, live provider or deployed sleep claim. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 29b46cddee7..1dad4ca398a 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -91,7 +91,8 @@ describe("createChatReconciliationCoordinator", () => { 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("); + expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications("); + expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,"); }); 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 29d1b0db701..59856338ce7 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,6 @@ +import { environmentService } from "../services/environments.js"; +import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; +import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js"; 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"; @@ -14,7 +17,7 @@ import { chatRunSourceMessageId } from "../services/chat-channels/run-origin.js" import * as durableChatWakeup from "../services/durable-chat-wakeup.js"; import { conflict } from "../errors.js"; import * as githubTaskPreparation from "../services/chat-channels/github/task-preparation.js"; -import { configureAgentLifecycle } from "../services/agent-lifecycle.js"; +import { configureAgentLifecycle, createAgentLifecycle } from "../services/agent-lifecycle.js"; import { budgetPolicies } from "@paperclipai/db"; import { budgetService } from "../services/budgets.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -28,7 +31,7 @@ import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@ import { githubResponseAuthorityFields } from "../services/chat-channels/github/response-authority-read.js"; import * as activityLogModule from "../services/activity-log.js"; import * as connectionCredentialBindings from "../services/connection-credential-bindings.js"; -import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates, environments } from "@paperclipai/db"; +import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates } from "@paperclipai/db"; import type { PaperclipCloudConnector, SealedConnectorCredentials } from "../services/paperclip-cloud-connector.js"; import { buildSlackAppManifest, SLACK_MANAGER_SCOPES } from "@paperclipai/shared"; import { registerChatActionWork, dueChatAction } from "../services/chat-action-work.js"; @@ -143,6 +146,7 @@ import { nativeRunResults, nativeRunFinalizations, environmentLeases, + environments, principalPermissionGrants, projects, projectWorkspaces, @@ -1098,17 +1102,18 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { secretsTmpDir, "master.key", ); - if (externalTestDatabaseUrl) { - db = createDb(externalTestDatabaseUrl); - } else { - tempDb = await startEmbeddedPostgresTestDatabase( - "paperclip-chat-channels-", - ); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-chat-channels-"); } + // Scheduler tests advance virtual time while PostgreSQL uses real I/O. + // Do not expire or time out its connections on those virtual clock jumps. + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, { + idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0, + }); }, 90_000); afterAll(async () => { + await db?.$client.end(); await tempDb?.cleanup(); if (previousKeyFile === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE; @@ -61698,45 +61703,6 @@ 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[] = []; @@ -64040,6 +64006,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { "integrity", "cleanup", "active_environment", + "notice_cleanup", + "notice_replacement", "live_pid", "missing_input", "late_provider_event", @@ -64073,6 +64041,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { : providerAccount; const previousStateDirectory = process.env.PAPERCLIP_RUNNER_STATE_DIR; let stateDirectory: string | null = null; + let noticeCoordinator: ReturnType | undefined; + let noticeActionId: string | undefined; + let noticeEnvironmentId: string | undefined; try { const nativeSessionId = randomUUID(); const runnerInstanceId = randomUUID(); @@ -64281,7 +64252,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); // A null account without existing provider-session evidence remains // conservatively ineligible for the exhausted-run recovery lane. - if (["bootstrap", "checkpoint", "distinct_account"].includes(kind)) { + if (["bootstrap", "checkpoint", "distinct_account", "notice_cleanup", "notice_replacement"].includes(kind)) { const staged = await attempt; await expect( context.service.processFailedChatRunRetry(staged.actionId), @@ -64291,12 +64262,101 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { retryOfRunId: context.runId, wakeCommentIds: [context.action.payload.commentId], }); + if (kind === "notice_cleanup" || kind === "notice_replacement") { + noticeActionId = staged.actionId; + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }) + .where(eq(agentWakeupRequests.id, staged.actionId)); + const cleanupAttemptId = randomUUID(); + if (kind === "notice_replacement") { + const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "ssh" }).returning(); + noticeEnvironmentId = environment.id; + } + const [lease] = await db.insert(environmentLeases).values({ + companyId: context.fixture.companyId, issueId: context.issue.id, + heartbeatRunId: context.runId, status: "pending_cleanup", cleanupStatus: "failed", + metadata: { pendingCleanupAttemptId: cleanupAttemptId, agentId: context.fixture.assignedAgentId }, + ...(kind === "notice_replacement" ? { environmentId: noticeEnvironmentId, + status: "retained", leasePolicy: "reuse_by_environment", providerLeaseId: randomUUID() } : {}), + }).returning(); + const errors: unknown[] = []; + noticeCoordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await chatFailedRetryNoticeWork(context.service).register(noticeCoordinator, () => true).ready; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); + expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + const replace = (owner: Db, replacesReusableLeaseId = lease.id) => environmentService(owner).acquireLease({ + companyId: context.fixture.companyId, environmentId: noticeEnvironmentId!, issueId: context.issue.id, + leasePolicy: "reuse_by_environment", providerLeaseId: lease.providerLeaseId, + metadata: { agentId: context.fixture.assignedAgentId }, replacesReusableLeaseId, + }); + if (kind === "notice_replacement") { + await expect(replace(db, randomUUID())).rejects.toThrow("ownership changed"); + const rollback = new Error("rollback reusable lease handoff"); + await expect(db.transaction(async tx => { + await replace(tx as unknown as Db); + throw rollback; + })).rejects.toBe(rollback); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); + expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await db.transaction(tx => replace(tx as unknown as Db)); + } else { + expect(await environmentService(db).releaseLease(lease.id, "released", { + cleanupStatus: "success", expectedPendingCleanupAttemptId: randomUUID(), + })).toBeNull(); + const rollback = new Error("rollback native lease settlement"); + await expect(db.transaction(async tx => { + await environmentService(tx as unknown as Db).releaseLease(lease.id, "released", { + cleanupStatus: "success", expectedPendingCleanupAttemptId: cleanupAttemptId, + }); + throw rollback; + })).rejects.toBe(rollback); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); + expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await db.transaction(async tx => { + await environmentService(tx as unknown as Db).releaseLease(lease.id, "released", { + cleanupStatus: "success", expectedPendingCleanupAttemptId: cleanupAttemptId, + }); + }); + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + const selection = vi.spyOn(context.service, "loadFailedRetryNoticeScopes"); + try { + if (kind === "notice_replacement") await expect(replace(db)).rejects.toThrow("ownership changed"); + else await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" }); + await vi.advanceTimersByTimeAsync(1000); + expect(selection).not.toHaveBeenCalled(); + } finally { selection.mockRestore(); } + 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([]); + } } else await expect(attempt).rejects.toMatchObject({ details: { code: "chat_failed_run_retry_not_authorized" }, }); } finally { + await noticeCoordinator?.stop(); + if (kind === "notice_cleanup" || kind === "notice_replacement") vi.useRealTimers(); await context.service.shutdown(); + if (noticeEnvironmentId) { + await db.delete(environmentLeases).where(eq(environmentLeases.environmentId, noticeEnvironmentId)); + await db.delete(environments).where(eq(environments.id, noticeEnvironmentId)); + } + if (noticeActionId) { + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, noticeActionId)); + await db.delete(chatActions).where(eq(chatActions.id, noticeActionId)); + } if (stateDirectory) { if (previousStateDirectory === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; @@ -64781,6 +64841,447 @@ 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.each(["normal", "drain", "failed selection", "stopped", "disabled"] as const)("reactivates failed retry notices from real admission after idle through %s", async mode => { + const context = await deferredFailedRetryNoticeFixture(mode !== "disabled"); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const selection = vi.spyOn(context.service, "loadFailedRetryNoticeScopes"); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const quiet = async () => { + const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled(); + } finally { select.mockRestore(); transaction.mockRestore(); } + }; + try { + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => canRun).ready; + await settle(); await quiet(); + if (mode === "drain") canRun = false; + if (mode === "stopped") await coordinator.stop(); + if (mode === "failed selection") selection.mockRejectedValueOnce(new Error("temporary retry selection failure")); + expect(await context.admit()).toMatchObject({ status: mode === "disabled" ? "failed" : "deferred" }); + if (mode === "failed selection") { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(errors).toHaveLength(1); }); + canRun = false; + } + if (["drain", "failed selection", "stopped"].includes(mode)) { + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + await quiet(); canRun = true; + if (mode === "stopped") { + await context.service.processPublicationMaintenance(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => canRun).ready; + } + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + payload: mode === "disabled" + ? { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" } + : { text: "Your retry is queued.", progressState: "queued" }, + })]); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + await quiet(); + expect(errors).toEqual(mode === "failed selection" ? [expect.objectContaining({ errors: [expect.objectContaining({ message: "temporary retry selection failure" })] })] : []); + } finally { + await coordinator.stop(); selection.mockRestore(); vi.useRealTimers(); await context.service.shutdown(); + await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, context.fixture.companyId), + inArray(agentWakeupRequests.id, [context.retry.requestId!, context.retry.chatActionId!]), isNull(agentWakeupRequests.runId))); + } + }); + + it.each(["idle", "bounded pass", "requester merge", "agent resume"] as const)("reactivates denied retry notices when authority is restored during %s", async mode => { + const context = await deferredFailedRetryNoticeFixture(); + const rejectedIds = mode === "bounded pass" ? 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; + let lifecycle: ReturnType | undefined; + const reconcile = context.service.reconcileFailedRetryNotices.bind(context.service); + const pass = vi.spyOn(context.service, "reconcileFailedRetryNotices"); + try { + expect(await context.admit()).toMatchObject({ status: "deferred" }); + const [originalReceipt] = await readAgentWakeupRequestsById(context.retry.chatActionId!); + if (mode === "agent resume") { + lifecycle = configureAgentLifecycle(db, { requiredPluginIds: async () => [], + runHost: async () => "complete", runPlugin: async () => "complete" }); + await createAgentLifecycle(db).pauseAgent(context.fixture.assignedAgentId); + await lifecycle.process(context.fixture.assignedAgentId); + expect((await db.select().from(agents).where(eq(agents.id, context.fixture.assignedAgentId)))[0]) + .toMatchObject({ status: "paused", lifecycleState: "paused" }); + } else if (mode === "requester merge") { + expect(originalReceipt.requestedByActorType).toBe("user"); + await db.update(agentWakeupRequests).set({ requestedByActorId: "another-requester" }) + .where(eq(agentWakeupRequests.id, originalReceipt.id)); + } else { + await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.issue.id, + createdByUserId: null, responsibleUserId: null, assigneeUserId: null, + }).where(eq(issues.id, context.issue.id)); + } + if (rejectedIds.length) { + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, context.retry.chatActionId!)); + const [receipt] = await readAgentWakeupRequestsById(action.id); + await db.insert(chatActions).values(rejectedIds.map(id => ({ ...action, id, + providerActionId: `retry-cursor:${id}`, createdAt: new Date(action.createdAt.getTime() + 1000), + payload: { ...action.payload, sourceScopeSha256: "invalid-source" }, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ ...receipt, id, + idempotencyKey: `retry-cursor:${id}`, + }))); + pass.mockImplementationOnce(async input => { + const result = await reconcile(input); + expect(result).toMatchObject({ inserted: 0, nextCursor: expect.any(Object) }); + held = true; await pending; return result; + }); + } + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + ready = chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready; + if (mode === "bounded pass") await vi.waitFor(() => expect(held).toBe(true), { timeout: 10_000 }); + else { + await ready; + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(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(); } + } + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0); + if (mode === "agent resume") { + await createAgentLifecycle(db).resumeAgent(context.fixture.assignedAgentId); + await lifecycle!.process(context.fixture.assignedAgentId); + expect((await db.select().from(agents).where(eq(agents.id, context.fixture.assignedAgentId)))[0]) + .toMatchObject({ status: "idle", lifecycleState: "ready" }); + } else if (mode === "requester merge") { + await db.transaction(async tx => { + await tx.execute(sql`select id from issues where id = ${context.issue.id} for update`); + await createWakeAdmissionWriter().mergeIntoExistingDeferredWake( + createAdmissionTransactionScope(context.fixture.companyId, tx as unknown as Db), { + companyId: context.fixture.companyId, existingDeferredWakeId: originalReceipt.id, + mergedPayload: originalReceipt.payload ?? {}, nextCoalescedCount: 1, + manualUserWakeActorId: originalReceipt.requestedByActorId!, + }); + }); + } else await issueService(db).update(context.issue.id, { visibility: "open" }); + release(); await ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + 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 lifecycle?.stop(); await coordinator.stop(); pass.mockRestore(); vi.useRealTimers(); + await context.service.shutdown(); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, + [...rejectedIds, context.retry.requestId!, context.retry.chatActionId!])); + if (rejectedIds.length) await db.delete(chatActions).where(inArray(chatActions.id, rejectedIds)); + } + }); + + it("wakes a not-started retry notice after resumed admission is denied", async () => { + const context = await deferredFailedRetryNoticeFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + try { + expect(await context.admit()).toMatchObject({ status: "deferred" }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).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); + }); + await reconcileTaskChatRetryReceipts(db); + context.wakeup.mockRejectedValueOnce(Object.assign(new Error("denied at resume"), { + details: { code: "chat_failed_run_retry_not_authorized" }, + })); + expect(await context.admit()).toMatchObject({ status: "failed" }); + expect((await readAgentWakeupRequestsById(context.retry.chatActionId!))[0].status).toBe("failed"); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(2); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + 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(); + await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, context.fixture.companyId), + inArray(agentWakeupRequests.id, [context.retry.requestId!, context.retry.chatActionId!]), isNull(agentWakeupRequests.runId))); + } + }); + + it.each(["agent", "tree"] as const)("wakes a failed retry notice only after %s cancellation commits", async kind => { + const context = await failedChatRetryFixture("telegram"); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + canReconcile: () => true, onError: error => errors.push(error) }); + let stagedActionId: string | undefined; + const settle = async () => vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + const cancel = (database: typeof db) => kind === "agent" + ? heartbeatService(database).cancelInvocationsForAgents([context.fixture.assignedAgentId], "Cancelled retry fixture") + : issueTreeControlService(database).cancelUnclaimedWakeupsForTree(context.fixture.companyId, context.issue.id, "Cancelled retry fixture"); + try { + const staged = await db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, { + companyId: context.fixture.companyId, issueId: context.issue.id, + agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user", + })); + stagedActionId = staged.actionId; + await context.service.processFailedChatRunRetry(staged.actionId); + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }) + .where(eq(agentWakeupRequests.id, staged.actionId)); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready; + await settle(); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + const rollback = new Error("rollback retry cancellation"); + await expect(db.transaction(async tx => { await cancel(tx as unknown as typeof db); throw rollback; })).rejects.toBe(rollback); + await settle(); + expect((await readAgentWakeupRequestsById(staged.actionId))[0].status).toBe("deferred_issue_execution"); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1); + await db.transaction(async tx => { await cancel(tx as unknown as typeof db); }); + await settle(); + expect((await readAgentWakeupRequestsById(staged.actionId))[0].status).toBe("cancelled"); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" }, + })])); + 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(); + if (stagedActionId) { + await db.delete(chatActions).where(eq(chatActions.id, stagedActionId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, stagedActionId)); + } + } + }); + + it.each(["manual", "scoped"] as const)("advances failed retry notices past denied history with microsecond timestamps (%s)", async mode => { + const context = await failedChatRetryFixture("telegram"); + const rejectedIds = Array.from({ length: 50 }, () => randomUUID()); + let stagedActionId: string | undefined; + try { + const staged = await db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, { + companyId: context.fixture.companyId, issueId: context.issue.id, + agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user", + })); + stagedActionId = staged.actionId; + await context.service.processFailedChatRunRetry(staged.actionId); + const [receipt] = await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }) + .where(eq(agentWakeupRequests.id, staged.actionId)).returning(); + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, staged.actionId)); + const before = new Date(action.createdAt.getTime() - 1000).toISOString(); + await db.insert(chatActions).values(rejectedIds.map(id => ({ + ...action, id, providerActionId: `failed-run-retry:${id}`, + payload: { ...action.payload, failedRunId: randomUUID() }, + createdAt: sql`${before}::timestamptz + interval '0.000123 seconds'`, + }))); + await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ + ...receipt, id, idempotencyKey: `retry-notice-cursor:${id}`, + }))); + if (mode === "manual") { + expect(await context.service.enqueueInboundWakeupPublications(50)).toBe(0); + expect(await context.service.enqueueInboundWakeupPublications(50)).toBe(1); + } else { + const scope = { companyId: context.fixture.companyId, issueId: context.issue.id }; + const transaction = vi.spyOn(db, "transaction"); + try { + const first = await context.service.reconcileFailedRetryNotices({ scope, limit: 50 }); + expect(first).toMatchObject({ inserted: 0, contended: false, nextCursor: expect.any(Object) }); + expect(transaction).toHaveBeenCalledTimes(50); transaction.mockClear(); + expect(await context.service.reconcileFailedRetryNotices({ scope, limit: 50, cursor: first.nextCursor })) + .toEqual({ inserted: 1, contended: false, nextCursor: null }); + expect(transaction).toHaveBeenCalledTimes(1); + } finally { transaction.mockRestore(); } + } + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({ + idempotencyKey: `wake:${staged.actionId}:queued:${context.endpoint.id}:${context.conversation.id}`, + payload: { text: "Your retry is queued.", progressState: "queued" }, + })]); + } finally { + await context.service.shutdown(); + const actionIds = stagedActionId ? [...rejectedIds, stagedActionId] : rejectedIds; + await db.delete(chatActions).where(inArray(chatActions.id, actionIds)); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, actionIds)); + } + }); + + it.each(["task", "company", "startup"] as const)("bounds failed retry notice scope discovery with retained candidates (%s)", async mode => { + const context = await failedChatRetryFixture("telegram"); + const other = await failedChatRetryFixture("telegram"); + const copiedIds = Array.from({ length: 250 }, () => randomUUID()); + const stagedIds: string[] = []; + try { + for (const target of [context, other]) { + const staged = await db.transaction(tx => target.service.prepareFailedChatRunRetry(tx, { + companyId: target.fixture.companyId, issueId: target.issue.id, + agentId: target.fixture.assignedAgentId, failedRunId: target.runId, initiatedByUserId: "owner-user", + })); + await target.service.processFailedChatRunRetry(staged.actionId); + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }) + .where(eq(agentWakeupRequests.id, staged.actionId)); + stagedIds.push(staged.actionId); + } + const [action] = await db.select().from(chatActions).where(eq(chatActions.id, stagedIds[0])); + const [receipt] = await readAgentWakeupRequestsById(stagedIds[0]); + await db.insert(chatActions).values(copiedIds.map(id => ({ + ...action, id, providerActionId: `failed_run_retry:${id}`, + }))); + await db.insert(agentWakeupRequests).values(copiedIds.map(id => ({ + ...receipt, id, idempotencyKey: `retry-discovery-budget:${id}`, + }))); + const scope = { companyId: context.fixture.companyId, issueId: context.issue.id }; + 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.loadFailedRetryNoticeScopes( + 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.issue.id, + }])); + } 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, copiedIds)); + await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, copiedIds)); + } + }); + + it("scopes failed retry notices and retains receipt contention without retrying settled history", async () => { + const context = await failedChatRetryFixture("telegram"); + const other = await failedChatRetryFixture("telegram"); + let release = () => {}; + let holder: Promise | undefined; + try { + const staged = []; + for (const target of [context, other]) { + const action = await db.transaction(tx => target.service.prepareFailedChatRunRetry(tx, { + companyId: target.fixture.companyId, issueId: target.issue.id, + agentId: target.fixture.assignedAgentId, failedRunId: target.runId, initiatedByUserId: "owner-user", + })); + await target.service.processFailedChatRunRetry(action.actionId); + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }) + .where(eq(agentWakeupRequests.id, action.actionId)); + staged.push(action); + } + const scope = { companyId: context.fixture.companyId, issueId: context.issue.id }; + expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([scope]); + 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, staged[0].actionId)).for("update"); + locked(); await hold; + }); + await ready; + expect(await context.service.reconcileFailedRetryNotices({ scope })) + .toEqual({ inserted: 0, contended: true, nextCursor: null }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]); + release(); await holder; + expect(await context.service.reconcileFailedRetryNotices({ scope })) + .toEqual({ inserted: 1, contended: false, nextCursor: null }); + expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([]); + expect(await readChatPublicationsByEndpointId(other.endpoint.id)).toEqual([]); + await db.update(agentWakeupRequests).set({ status: "cancelled" }) + .where(eq(agentWakeupRequests.id, staged[0].actionId)); + expect(await context.service.reconcileFailedRetryNotices({ scope })) + .toEqual({ inserted: 1, contended: false, nextCursor: null }); + expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({ + payload: { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" }, + })])); + expect(await context.service.reconcileFailedRetryNotices({ scope })) + .toEqual({ inserted: 0, contended: false, nextCursor: null }); + expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([]); + await db.update(agentWakeupRequests).set({ runId: other.runId }) + .where(eq(agentWakeupRequests.id, staged[1].actionId)); + expect(await other.service.loadFailedRetryNoticeScopes({ companyId: other.fixture.companyId })).toEqual([]); + } finally { release(); await holder; await context.service.shutdown(); await other.service.shutdown(); } + }); + it.each(["slack", "telegram"] as const)( "keeps a %s retry's queue, working, and selected final on its own provider identity", async (provider) => { diff --git a/server/src/__tests__/environment-service.test.ts b/server/src/__tests__/environment-service.test.ts index 59774e35ec5..f1a1c98f05c 100644 --- a/server/src/__tests__/environment-service.test.ts +++ b/server/src/__tests__/environment-service.test.ts @@ -11,6 +11,7 @@ import { companySecretBindings, companySecrets, createDb, + closeRegisteredClients, environmentCustomImageSetupSessions, environmentLeases, environments, @@ -26,7 +27,11 @@ import { } from "./helpers/embedded-postgres.js"; import { environmentService } from "../services/environments.ts"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL; +if (externalTestDatabaseUrl && !new URL(externalTestDatabaseUrl).pathname.endsWith("_test")) { + throw new Error("Environment service tests require an isolated database ending in _test"); +} +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -41,9 +46,14 @@ describeEmbeddedPostgres("environmentService leases", () => { let svc!: ReturnType; beforeAll(async () => { - const started = await startEmbeddedPostgresTestDatabase("environment-service"); - stopDb = started.stop; - db = createDb(started.connectionString); + if (externalTestDatabaseUrl) { + db = createDb(externalTestDatabaseUrl); + stopDb = () => closeRegisteredClients(externalTestDatabaseUrl); + } else { + const started = await startEmbeddedPostgresTestDatabase("environment-service"); + stopDb = started.stop; + db = createDb(started.connectionString); + } svc = environmentService(db); }); diff --git a/server/src/app.ts b/server/src/app.ts index ffb2e212678..09ac7213ef6 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -197,6 +197,7 @@ 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 { chatFailedRetryNoticeWork } from "./services/chat-failed-retry-notice-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; import { createCoalescedAsyncTrigger, @@ -1290,6 +1291,7 @@ export async function createApp( chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatFailedRetryNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatRunMilestoneWork(db, opts.authPublicBaseUrl).register( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); diff --git a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts index f6c92607fc5..246cb9f707a 100644 --- a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts @@ -1,3 +1,6 @@ +import { subscribeDatabaseWork } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.js"; +import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; import { createAgentLifecycle as createLifecycleCommands } from "../index.js"; import { createAgentLifecycleEffects } from "../../../services/agent-lifecycle.js"; import { deleteCompany } from "../../../services/company-deletion.js"; @@ -26,14 +29,13 @@ import { transition } from "../domain/policy.js"; import type { LifecycleAgent } from "../application/ports.js"; import { budgetService } from "../../../services/budgets.js"; import { agentConfigurationRevision } from "../../../lib/agent-configuration-revision.js"; -import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js"; const createLifecycleStore = (db: Db) => createStore(db, createAgentLifecycleEffects()); const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_LIFECYCLE_TEST_DATABASE_URL"); const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); (support.supported ? describe : describe.skip)("agent lifecycle commands", () => { - let database: Awaited>; + let database: Awaited> | undefined; let db: Db; let companyId: string; const workers: ReturnType[] = []; @@ -637,6 +639,48 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu } }); + it("settles notice reactivation only with the outer lifecycle transaction", async () => { + const work = worker(); + const agent = await hire(); + await work.process(agent.id); + await createAgentLifecycle(db).pauseAgent(agent.id); + await work.process(agent.id); + await work.stop(); + await createAgentLifecycle(db).resumeAgent(agent.id); + const store = createLifecycleStore(db); + const owner = "resume-notice-test"; + const claimed = await store.claim(agent.id, owner, new Date()); + expect(claimed).not.toBeNull(); + await store.setRequiredPlugins(claimed!, owner, []); + await store.completeHost(claimed!, owner); + const events: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatFailedRetryNotices, (event, scopes) => { + if ([...scopes ?? []].some(scope => JSON.parse(scope).companyId === companyId)) events.push(event); + }); + try { + await expect(db.transaction(async tx => { + expect(await createLifecycleStore(tx as unknown as Db).change(agent.id, "complete", { owner })) + .toMatchObject({ status: "idle", lifecycleState: "ready" }); + expect(events).toEqual(["intent"]); + expect(await current(agent.id)).toMatchObject({ status: "paused", lifecycleState: "resuming" }); + throw new Error("rollback resumed agent"); + })).rejects.toThrow("rollback resumed agent"); + expect(events).toEqual(["intent", "settled"]); + expect(await current(agent.id)).toMatchObject({ status: "paused", lifecycleState: "resuming" }); + events.length = 0; + await db.transaction(async tx => { + await createLifecycleStore(tx as unknown as Db).change(agent.id, "complete", { owner }); + expect(events).toEqual(["intent"]); + expect((await current(agent.id)).status).toBe("paused"); + }); + expect(events).toEqual(["intent", "settled"]); + expect(await current(agent.id)).toMatchObject({ status: "idle", lifecycleState: "ready" }); + events.length = 0; + await store.change(agent.id, "reconcile"); + expect(events).toEqual([]); + } finally { unsubscribe(); } + }); + it("rejects transaction injection and generic state writes", async () => { const agent = await hire("pending_approval"); await expect(db.transaction(async tx => createAgentLifecycle(tx as unknown as Db))).rejects.toThrow("root database"); diff --git a/server/src/modules/agent-lifecycle/adapters/effects.ts b/server/src/modules/agent-lifecycle/adapters/effects.ts index e2fd735b1db..57de2cc5468 100644 --- a/server/src/modules/agent-lifecycle/adapters/effects.ts +++ b/server/src/modules/agent-lifecycle/adapters/effects.ts @@ -8,6 +8,7 @@ import type { ActivityPublication } from "../../../types/activity-publication.js export interface LifecycleEffects { setupRequiresPreparation(db: Db, companyId: string, environmentId?: string | null): Promise; notifyWork(db: Db): Promise; + beforeStatusChange(db: Db, companyId: string, agentId: string, before: string, after: string): Promise; reportFailure(context: LifecycleFailure, error: unknown): void; deleteDependencies(db: Db, companyId: string, agentId: string): Promise; prepareConfiguration(db: Db, current: AgentConfigurationRecord, data: AgentConfigurationPatch, options?: UpdateAgentOptions): Promise; diff --git a/server/src/modules/agent-lifecycle/adapters/postgres.ts b/server/src/modules/agent-lifecycle/adapters/postgres.ts index 4179fa923aa..0e618f2ae5f 100644 --- a/server/src/modules/agent-lifecycle/adapters/postgres.ts +++ b/server/src/modules/agent-lifecycle/adapters/postgres.ts @@ -80,6 +80,7 @@ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: Lifec }; if (next === "ready") operation.resumeState = "ready"; const status = compatibilityStatus(next, agent.status); + if (status !== agent.status) await effects.beforeStatusChange(tx, agent.companyId, agent.id, agent.status, status); if (isLifecycleWorkPending(next)) await effects.notifyWork(tx); const [updated] = await tx.update(agents).set({ lifecycleHolds: holds, lifecycleState: next, lifecycleVersion: agent.lifecycleVersion + 1, lifecycleError: null, diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index f76cbacb844..078a3d04aa0 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -996,6 +996,7 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter { async mergeIntoExistingDeferredWake(scope, input) { const tx = requireAdmissionTx(scope, input.companyId); + if (input.manualUserWakeActorId) await notifyChatInboundWakeNotices(tx, { companyId: input.companyId }); const rows = await tx .update(agentWakeupRequests) .set({ diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index d5b9346e508..164e94420db 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,5 @@ +import { isAgentStatusInvokable } from "@paperclipai/shared"; +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { logger } from "../middleware/logger.js"; import { companies, type Db } from "@paperclipai/db"; import { eq } from "drizzle-orm"; @@ -33,6 +35,11 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li return true; }, notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle), + async beforeStatusChange(tx, companyId, _agentId, before, after) { + if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { + await notifyChatInboundWakeNotices(tx, { companyId }); + } + }, reportFailure(context, error) { const code = lifecycle.lifecycleFailureCode(error); logger.warn({ ...context, failureCode: code ?? "unclassified" }, "Agent lifecycle operation failed"); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index f08bc42cba4..1675fd413e0 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1236,6 +1236,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { processFailedChatRunRetry, processFailedChatRunRetries, enqueueFailedChatRetryPublications, + loadFailedRetryNoticeScopes, + reconcileFailedRetryNotices, pendingFailedRetries, } = createChatFailedRunRetry(db, { authorizeInboundWakeup, @@ -3348,9 +3350,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPublicationMaintenance(limit = 25) { scheduleGitHubMaintenance(); - // 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); } @@ -3504,6 +3503,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + loadFailedRetryNoticeScopes, + reconcileFailedRetryNotices, registerTeamsAttemptRecoveryWork, registerTeamsConsentExpiryWork, registerCommandWork, diff --git a/server/src/services/chat-channels/failed-retry-notices.ts b/server/src/services/chat-channels/failed-retry-notices.ts new file mode 100644 index 00000000000..effab8776ef --- /dev/null +++ b/server/src/services/chat-channels/failed-retry-notices.ts @@ -0,0 +1,107 @@ +import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm"; +import { agentWakeupRequests, chatActions, chatPublications, type Db } from "@paperclipai/db"; +import type { DbOrTransaction } from "./runtime-types.js"; +import { notifyChatPublicationWork } from "../chat-work-notifications.js"; +import { projectSafeChatPublication } from "../chat-publication-projection.js"; +import { inboundWakePublicationKey } from "../chat-inbound-wakeup-publications.js"; +import { NativeChatReviewPresentationContentionError } from "../native-runtime/native-chat-review-presentation.js"; +import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js"; + +export type FailedRetryNoticeScope = { companyId: string; issueId?: string }; +export type FailedRetryNoticeCursor = { createdAt: string; id: string }; +type Options = { + authorize: (tx: DbOrTransaction, action: typeof chatActions.$inferSelect) => Promise<{ + endpointId: string; conversationId: string; issueId: string; commentIds: string[]; + }>; + isExternalActionAuthorizationChange: (error: unknown) => boolean; + isShuttingDown: () => boolean; +}; + +export function createFailedRetryNotices(db: Db, options: Options) { + let manualCursor: FailedRetryNoticeCursor | null = null; + function candidateQuery(cursor: FailedRetryNoticeCursor | null, scope?: FailedRetryNoticeScope) { + return db.select({ action: chatActions, receipt: agentWakeupRequests, + cursorCreatedAt: sql`${chatActions.createdAt}::text`, + companyId: sql`${chatActions.companyId}`.as("retry_notice_company_id"), + issueId: sql`${chatActions.payload}->>'issueId'`.as("retry_notice_issue_id"), + }).from(chatActions).innerJoin(agentWakeupRequests, and( + eq(agentWakeupRequests.id, chatActions.id), eq(agentWakeupRequests.companyId, chatActions.companyId), + )).where(and( + eq(chatActions.kind, "failed_run_retry"), isNull(agentWakeupRequests.runId), + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? sql`${chatActions.payload}->>'issueId' = ${scope.issueId}` : undefined, + sql`${chatActions.payload}->>'issueId' ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'`, + cursor ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt}::timestamptz, ${cursor.id}::uuid)` : undefined, + inArray(agentWakeupRequests.status, ["deferred_issue_execution", "cancelled", "failed", "skipped"]), + sql`not exists (select 1 from chat_publications retry_notice where retry_notice.company_id = ${chatActions.companyId} + and retry_notice.idempotency_key = 'wake:' || ${chatActions.id}::text || ':' || + case when ${agentWakeupRequests.status} = 'deferred_issue_execution' then 'queued' else 'not_started' end || ':' || + ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`, + )); + } + async function candidates(limit: number, cursor: FailedRetryNoticeCursor | null, scope?: FailedRetryNoticeScope) { + return candidateQuery(cursor, scope).orderBy(asc(chatActions.createdAt), asc(chatActions.id)).limit(limit); + } + type Candidates = Awaited>; + function continuation(rows: Candidates, limit: number): FailedRetryNoticeCursor | null { + const last = rows.at(-1); + return rows.length === limit && last ? { createdAt: last.cursorCreatedAt, id: last.action.id } : null; + } + async function project(rows: Candidates, signal?: AbortSignal) { + let inserted = 0, contended = false; + for (const { action, receipt } of rows) { + signal?.throwIfAborted(); + if (options.isShuttingDown()) throw new Error("Failed retry notice reconciliation stopped"); + try { + inserted += await db.transaction(async tx => { + const source = await options.authorize(tx, action); + const [current] = await tx.select().from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, receipt.id)).for("update", { noWait: true }); + if (!current || current.runId || current.status !== receipt.status) return 0; + const state = current.status === "deferred_issue_execution" ? "queued" : "not_started"; + await notifyChatPublicationWork(tx); + return (await tx.insert(chatPublications).values({ + companyId: action.companyId, endpointId: source.endpointId, conversationId: source.conversationId, + issueId: source.issueId, commentId: source.commentIds.at(-1)!, + idempotencyKey: inboundWakePublicationKey(action.id, state, source.endpointId, source.conversationId), + payload: projectSafeChatPublication({ classification: "external", source: "safe_milestone", + text: state === "queued" ? "Your retry is queued." + : "This retry was not started. Open the task in Paperclip for details.", + progressState: state === "queued" ? "queued" : "failed", + }), state: "pending", + }).onConflictDoNothing().returning({ id: chatPublications.id })).length; + }); + } catch (error) { + if (isExternalChatWaitAuthorizationContention(error) + || error instanceof NativeChatReviewPresentationContentionError) contended = true; + else if (options.isExternalActionAuthorizationChange(error) + || (error && typeof error === "object" && "status" in error && error.status === 409)) continue; + else throw error; + } + } + return { inserted, contended }; + } + async function enqueueFailedChatRetryPublications(limit: number) { + if (options.isShuttingDown()) return 0; + const size = Math.max(1, Math.min(50, limit)); + const rows = await candidates(size, manualCursor); + manualCursor = continuation(rows, size); + return (await project(rows)).inserted; + } + async function loadFailedRetryNoticeScopes(scope?: FailedRetryNoticeScope, signal?: AbortSignal) { + signal?.throwIfAborted(); + if (scope?.issueId) return [{ companyId: scope.companyId, issueId: scope.issueId }]; + const candidates = candidateQuery(null, scope).as("failed_retry_notice_candidates"); + return db.selectDistinct({ companyId: candidates.companyId, issueId: candidates.issueId }) + .from(candidates); + } + async function reconcileFailedRetryNotices(input: { + scope: FailedRetryNoticeScope; limit?: number; cursor?: FailedRetryNoticeCursor | null; signal?: AbortSignal; + }) { + input.signal?.throwIfAborted(); + const limit = Math.max(1, Math.min(input.limit ?? 100, 200)); + const rows = await candidates(limit, input.cursor ?? null, input.scope); + return { ...await project(rows, input.signal), nextCursor: continuation(rows, limit) }; + } + return { enqueueFailedChatRetryPublications, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices }; +} diff --git a/server/src/services/chat-channels/failed-run-retry.ts b/server/src/services/chat-channels/failed-run-retry.ts index 26b53a0e1e8..8dd6ace33a9 100644 --- a/server/src/services/chat-channels/failed-run-retry.ts +++ b/server/src/services/chat-channels/failed-run-retry.ts @@ -1,4 +1,5 @@ import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; +import { createFailedRetryNotices } from "./failed-retry-notices.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"; @@ -87,13 +88,8 @@ import { import { NativeChatReviewPresentationContentionError, } from "../native-runtime/native-chat-review-presentation.js"; -import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js"; import { CHAT_CONVERSATION_ID_KEY } from "../../modules/run-dispatch/index.js"; -import { projectSafeChatPublication } from "../chat-publication-projection.js"; -import { - inboundWakePublicationKey, -} from "../chat-inbound-wakeup-publications.js"; type DbOrTransaction = Db | Parameters[0]>[0]; type DeliveryRow = typeof chatDeliveries.$inferSelect; @@ -1980,117 +1976,13 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti return candidates.length; } - let failedRetryNoticeCursor: { createdAt: Date; id: string } | null = null; - async function enqueueFailedChatRetryPublications( - limit: number, - ): Promise { - const cursor = failedRetryNoticeCursor; - const batchSize = Math.max(1, Math.min(50, limit)); - const candidates = await db - .select({ action: chatActions, receipt: agentWakeupRequests }) - .from(chatActions) - .innerJoin( - agentWakeupRequests, - and( - eq(agentWakeupRequests.id, chatActions.id), - eq(agentWakeupRequests.companyId, chatActions.companyId), - ), - ) - .where( - and( - eq(chatActions.kind, "failed_run_retry"), - isNull(agentWakeupRequests.runId), - cursor - ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt.toISOString()}::timestamptz, ${cursor.id}::uuid)` - : undefined, - inArray(agentWakeupRequests.status, [ - "deferred_issue_execution", - "cancelled", - "failed", - "skipped", - ]), - sql`not exists (select 1 from chat_publications retry_notice where retry_notice.company_id = ${chatActions.companyId} - and retry_notice.idempotency_key = 'wake:' || ${chatActions.id}::text || ':' || - case when ${agentWakeupRequests.status} = 'deferred_issue_execution' then 'queued' else 'not_started' end || ':' || - ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`, - ), - ) - .orderBy(asc(chatActions.createdAt), asc(chatActions.id)) - .limit(batchSize); - const last = candidates.at(-1)?.action; - failedRetryNoticeCursor = - candidates.length >= batchSize && last - ? { createdAt: last.createdAt, id: last.id } - : null; - let inserted = 0; - for (const { action, receipt } of candidates) { - try { - inserted += await db.transaction(async (tx) => { - const source = await authorizeFailedChatRunRetry(tx, { - companyId: action.companyId, - issueId: String(action.payload.issueId), - agentId: String(action.payload.agentId), - retryActionId: action.id, - publication: true, - }); - const [current] = await tx - .select() - .from(agentWakeupRequests) - .where(eq(agentWakeupRequests.id, receipt.id)) - .for("update", { noWait: true }); - if (!current || current.runId || current.status !== receipt.status) - return 0; - const state = - current.status === "deferred_issue_execution" - ? "queued" - : "not_started"; - await notifyChatPublicationWork(tx); - return ( - await tx - .insert(chatPublications) - .values({ - companyId: action.companyId, - endpointId: source.endpointId, - conversationId: source.conversationId, - issueId: source.issueId, - commentId: source.commentIds.at(-1)!, - idempotencyKey: inboundWakePublicationKey( - action.id, - state, - source.endpointId, - source.conversationId, - ), - payload: projectSafeChatPublication({ - classification: "external", - source: "safe_milestone", - text: - state === "queued" - ? "Your retry is queued." - : "This retry was not started. Open the task in Paperclip for details.", - progressState: state === "queued" ? "queued" : "failed", - }), - state: "pending", - }) - .onConflictDoNothing() - .returning({ id: chatPublications.id }) - ).length; - }); - } catch (error) { - if ( - isExternalChatWaitAuthorizationContention(error) || - error instanceof NativeChatReviewPresentationContentionError || - isExternalActionAuthorizationChange(error) || - (error && - typeof error === "object" && - "status" in error && - error.status === 409) - ) - continue; - throw error; - } - } - return inserted; - } + const retryNotices = createFailedRetryNotices(db, { + authorize: (tx, action) => authorizeFailedChatRunRetry(tx, { + companyId: action.companyId, issueId: String(action.payload.issueId), + agentId: String(action.payload.agentId), retryActionId: action.id, publication: true, + }), + isExternalActionAuthorizationChange, isShuttingDown, + }); return { retryCommentIds, @@ -2103,7 +1995,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti unregisterFailedRetryAuthority, processFailedChatRunRetry, processFailedChatRunRetries, - enqueueFailedChatRetryPublications, + ...retryNotices, pendingFailedRetries: () => [...failedRetryTasks.values()], }; } diff --git a/server/src/services/chat-failed-retry-notice-work.ts b/server/src/services/chat-failed-retry-notice-work.ts new file mode 100644 index 00000000000..c9075eb294c --- /dev/null +++ b/server/src/services/chat-failed-retry-notice-work.ts @@ -0,0 +1,39 @@ +import type { ChatChannelService } from "./chat-channels.js"; +import type { FailedRetryNoticeCursor, FailedRetryNoticeScope } from "./chat-channels/failed-retry-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: FailedRetryNoticeCursor | null; contended: boolean; +}; +const batchSize = 100; + +export function chatFailedRetryNoticeWork(service: Pick) { + async function load(scope?: FailedRetryNoticeScope, signal?: AbortSignal): Promise { + return (await service.loadFailedRetryNoticeScopes(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.reconcileFailedRetryNotices({ + 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() }; + return contended ? { ...entry, cursor: null, contended: false, dueAt: Date.now() + 1000 } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatFailedRetryNotices, + decodeScope: (hint: string): FailedRetryNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: FailedRetryNoticeScope) => !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-inbound-wake-notice-notifications.ts b/server/src/services/chat-inbound-wake-notice-notifications.ts index 12e0aa08422..6389ffe8755 100644 --- a/server/src/services/chat-inbound-wake-notice-notifications.ts +++ b/server/src/services/chat-inbound-wake-notice-notifications.ts @@ -1,7 +1,8 @@ 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 })); +export async function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) { + const hint = JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices, hint); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatFailedRetryNotices, hint); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index d652eef12f2..444165f9894 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -59,6 +59,7 @@ export const DELIVERY_QUEUES = { chatPublications: "chat-publications", chatNativeProgress: "chat-native-progress", chatInboundWakeNotices: "chat-inbound-wake-notices", + chatFailedRetryNotices: "chat-failed-retry-notices", chatRunMilestones: "chat-run-milestones", chatReceipts: "chat-slack-receipts", browser: "browser-use", diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index ec5abac4cae..91a6153d5f3 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { writeEnvironmentCleanup, notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { randomUUID } from "node:crypto"; @@ -1549,6 +1550,7 @@ export function environmentService(db: Db) { } } if (input.replacesReusableLeaseId) { + await notifyChatInboundWakeNotices(tx, { companyId: input.companyId }); const retired = await tx .update(environmentLeases) .set({ @@ -1671,8 +1673,24 @@ export function environmentService(db: Db) { : sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`; const cleanupSettled = options?.expectedPendingCleanupAttemptId && options.cleanupStatus === "success" && status !== "pending_cleanup"; - const row = await writeEnvironmentCleanup(db, tx => tx - .update(environmentLeases) + const row = await writeEnvironmentCleanup(db, async tx => { + const condition = and(eq(environmentLeases.id, id), options?.expectedPendingCleanupAttemptId + ? and(eq(environmentLeases.status, "pending_cleanup"), + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.expectedPendingCleanupAttemptId}`) + : undefined); + const [previous] = await tx.select({ companyId: environmentLeases.companyId, issueId: environmentLeases.issueId, + heartbeatRunId: environmentLeases.heartbeatRunId, status: environmentLeases.status, + cleanupStatus: environmentLeases.cleanupStatus, releasedAt: environmentLeases.releasedAt, + }).from(environmentLeases).where(condition).for("update"); + if (!previous) return null; + if (previous.heartbeatRunId && (previous.status !== status || + (options?.cleanupStatus !== undefined && previous.cleanupStatus !== options.cleanupStatus) || + (status !== "retained" && !previous.releasedAt))) { + await notifyChatInboundWakeNotices(tx, { + companyId: previous.companyId, issueId: previous.issueId ?? undefined, + }); + } + return tx.update(environmentLeases) .set({ status, releasedAt: status === "retained" ? null : now, @@ -1686,12 +1704,10 @@ export function environmentService(db: Db) { ? sql`coalesce(${releaseMetadata}, '{}'::jsonb) || '{"pendingCleanupInFlight":false,"pendingCleanupLeaseExpiresAtMs":0}'::jsonb` : releaseMetadata, }) - .where(and(eq(environmentLeases.id, id), options?.expectedPendingCleanupAttemptId - ? and(eq(environmentLeases.status, "pending_cleanup"), - sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.expectedPendingCleanupAttemptId}`) - : undefined)) + .where(condition) .returning() - .then((rows) => rows[0] ?? null)); + .then((rows) => rows[0] ?? null); + }); return row ? toEnvironmentLease(row) : null; }, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 7778b17ac8b..580b46e298c 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { assertTaskRetryConversationAllowed, readTaskRetryIntent, readTaskRetryBoardMessageOrigin, retireSupersededTaskRetry, taskRetryMutationAllowed } from "../task-retry.js"; import { insertIssueComment } from "../issue-comment-writes.js"; @@ -2498,12 +2499,24 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const waitIssueId = issueId; return db.transaction(async (tx) => { await tx.execute(sql`select id from issues where id = ${waitIssueId} and company_id = ${agent.companyId} for update`); + if (durableRequest) await notifyChatInboundWakeNotices(tx, { + companyId: durableRequest.companyId, issueId: durableRequest.issueId, + }); return recordExecutionWait(tx as unknown as Db, { issueId: waitIssueId, request, condition: waitCondition, coalesce: coalesceExecutionWait, }); }); } - await db.insert(agentWakeupRequests).values(request); + if (durableRequest) { + await db.transaction(async tx => { + await notifyChatInboundWakeNotices(tx, { + companyId: durableRequest.companyId, issueId: durableRequest.issueId, + }); + await tx.insert(agentWakeupRequests).values(request); + }); + } else { + await db.insert(agentWakeupRequests).values(request); + } return { created: true }; }; const writeSkippedHeartbeatRequest = async (