From ba5bba86dcdde32b14d4d7dec1b51898e5f6f22d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:06:42 -0500 Subject: [PATCH] fix(server): wake GitHub acknowledgements from committed changes (#17656) https://entire.com/gh/paperclipai/-/trails/206/changes/paperclip/206 ## Thinking Path > - Paperclip coordinates agent work and presents its progress in chat. > - GitHub deliveries save acknowledgement actions and provider receipts. > - Periodic maintenance currently searches for unfinished acknowledgements even when idle. > - Committed delivery, lease, response and presentation changes identify when work can advance. > - The existing owner should wake on those changes and recover only at saved deadlines. > - Completed history should become quiet without losing unfinished provider effects. ## Linked Issues or Issue Description Refs #17653, which is merged. This existing GitHub acknowledgement slice targets master. **Problem:** A saved GitHub acknowledgement can wait for a publication lease, fast-response result or native presentation decision. Polling empty or unchanged work keeps the tenant active. Removing maintenance without covering these producers can strand the acknowledgement. **Expected behavior:** Committed changes wake the scoped owner. It reconstructs existing actions on startup, retries at saved deadlines, parks unchanged presentation waits and becomes quiet when settled. ## What Changed - Register GitHub acknowledgement work with the shared scoped deadline coordinator. Reuse existing action, delivery, lease and fast-response rows. - Add transaction-aware response-availability notifications at supported writes. Keep GitHub protocol details in its platform owner and shared notifications limited to IDs. - Share the bounded latest-request/run projection between scheduling and direct response authorization. An absent native presentation decision parks the action until its real writer commits. - Retain publication leases, source/principal authority, version fencing, response markers and final receipt precedence. Check admission before each new provider request; let in-flight work settle. Stop does not consume the provider retry budget. - Remove only recurring acknowledgement recovery and keep its manual path. Add no work table or scheduler service. - Regenerate the three original indexes as `0400_absent_phantom_reporter.sql`: pending acknowledgement IDs, fast-response delivery lookup and publication comment/destination lookup. Each uses concurrent/idempotent creation and the existing top-level migration runner. Preserve all landed SQL/journal entries and the five newest snapshots. ## Verification Published head: `09ca00b9f029838434ec042cda691345d67b493e`, on landed continuation parent `063cfd43813cabf18417605fb9eefb9aacf95c8d`. - **87 distinct current cases pass:** 18 real PostgreSQL acknowledgement lifecycle/presentation cases, all 26 fast-response service cases, all 30 heartbeat completion cases and all 13 startup cases. Database build, server typechecking/build, module boundaries and three fresh migrations through 0400 pass. The current Runner bundle rebuild also passes after upstream provider-event changes. - The lifecycle matrix covers empty/history quiet, lease release, restart, drain/standby, fast-response failure/expiry/publication/unknown outcome, streaming quarantine, outer rollback/commit, saved provider retry, stop between requests and receipt state. The real presentation writer proves quiet while pending, rollback, committed reactivation and final quiet. - The earlier **305-case** matrix (236 full GitHub workflow cases, 26 fast-response service cases, 30 completion cases and 13 startup cases) passes at `337f272853d4f8a51b5b9300900a46eb4e13f698`. It retains its prepared-parent scope. Repeated cases are not added to current 87. - A populated 0399 database upgrades to 0400 with all 398 prior migration receipts unchanged, one new receipt and no additional receipt on replay. All three indexes are valid/ready on fresh and upgraded databases. Their SQL definitions match the original draft after whitespace normalization; migration history and retained snapshot bytes are preserved. - Normal integration `9b3607e159153e00ccdfe538d52b591bd38d6499` preserves the full 28-path owned binary patch and all 20 unrelated master paths, including native retry, provider-event and evaluation workflow changes. Both normal merge parents and the exact tree are checked. No force push or rebase occurs. - [The owner audit](https://github.com/paperclipai/paperclip/blob/09ca00b9f029838434ec042cda691345d67b493e/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md#acceptance-on-the-actual-landed-continuation-parent) retains the original missing-notification, JSON precedence, receipt lifecycle, external-database fixture and migration-registry failures. Earlier 69/228/broad counts keep their original scope. Skips and failed revisions are not relabelled as current passes. - The final audit-only edit leaves all owned source/test/migration hashes unchanged. **Fresh exact-head CI and Greptile 5/5 with no actionable findings remain required.** No full local repository, live GitHub tenant or deployed sleep result is claimed. ## Risks - Process-local transaction hints cover supported application writers and shared pools. Independent SQL writers and other replicas do not gain a new wake guarantee. - Provider effects remain guarded by the existing publication lease and current source/principal authority. Unknown outcomes retain their existing recovery semantics. - Presentation waits depend on their supported milestone/transaction producers. A missing producer can defer recovery until another supported change or restart. - Index creation can take time on large histories and runs concurrently outside a transaction. Existing migration recovery handles replay; prior SQL is unchanged. - Worker quiet is distinct from host/database suspension. Cloud wake coverage and sleep admission remain unchanged; later maintenance duties are separate slices. ## Model Used OpenAI Codex, using reasoning, repository inspection, code execution and real PostgreSQL tests. The exact model identifier and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 43 ++ ...-25-github-chat-maintenance-sleep-audit.md | 409 ++++++++++++++++++ packages/db/src/client.ts | 1 + .../0400_absent_phantom_reporter.sql | 5 + ...{0395_snapshot.json => 0400_snapshot.json} | 182 +++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/chat_channels.ts | 5 + packages/db/src/schema/fast_responses.ts | 2 + .../app-private-hostname-gate.test.ts | 4 + .../chat-channels.integration.test.ts | 285 +++++++++++- server/src/__tests__/fast-responses.test.ts | 9 +- server/src/app.ts | 1 + server/src/services/chat-channels.ts | 2 + .../github/acknowledgement-work.ts | 112 +++++ .../chat-channels/github/maintenance.ts | 2 +- .../github/response-run-state.ts | 71 +++ .../chat-channels/publications/failures.ts | 2 + .../chat-channels/publications/queue.ts | 5 +- .../publications/task-control.ts | 2 + .../chat-channels/publications/worker.ts | 3 + .../services/chat-github-publication-lease.ts | 1 + .../services/chat-github-response-comments.ts | 141 +++--- .../services/chat-response-availability.ts | 7 + .../chat-run-milestone-notifications.ts | 3 + .../src/services/chat-work-notifications.ts | 2 + .../services/delivery-work-notifications.ts | 1 + server/src/services/fast-responses.ts | 35 +- .../src/services/heartbeat/run-completion.ts | 22 +- 28 files changed, 1253 insertions(+), 111 deletions(-) create mode 100644 doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md create mode 100644 packages/db/src/migrations/0400_absent_phantom_reporter.sql rename packages/db/src/migrations/meta/{0395_snapshot.json => 0400_snapshot.json} (99%) create mode 100644 server/src/services/chat-channels/github/acknowledgement-work.ts create mode 100644 server/src/services/chat-channels/github/response-run-state.ts create mode 100644 server/src/services/chat-response-availability.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 940f9cac89d..5a2d80fa0a4 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2326,3 +2326,46 @@ start another send after drain/standby/shutdown. Three regressions reproduced it. Per-send admission/abort checks now pass all 14 new lifecycle cases, and the previously skipped publication suite plus startup guards pass 27 cases with the external test database. Fresh corrected-head verification remains required. + +## Slice 33 accepted verification and GitHub owner split + +#17653 at `276ef23d2a0b01b15e6bfb8ff31ab13495741b19` passes all 53 +remote checks, fresh Greptile 5/5, and has no unresolved review threads. Workspace +typecheck/build pass. All 1,540 tests across chat integration, standalone +interaction publication, and startup guards passed on a fresh database with no +skips in 473.47 seconds. The initial skipped suite and the stop-boundary failure +remain documented. The PR is unmerged. + +Slice 34 starts from that verified head and owns initial GitHub response-comment +acknowledgement delivery. The same maintenance helper also contains terminal +response reconciliation, review-check publication, and late stale-head review +projection repair. Those are separate reasons and remain separate planned PRs. +See `2026-10-25-github-chat-maintenance-sleep-audit.md` for the inspected owners, +producer boundaries, and response-slot dependencies. No GitHub migration is yet +claimed implemented or verified. + +## Slice 34: initial GitHub acknowledgement delivery + +Based on accepted slice 33 (`276ef23d2a0b01b15e6bfb8ff31ab13495741b19`). +The implementation moves only initial response-comment acknowledgement out of +recurring GitHub maintenance. The existing response actions, fast-response +records, and publication leases own durability. Committed response-slot changes +and actual saved deadlines drive recovery. Terminal responses, review checks, +and stale-head projection remain separate upcoming reasons. + +Fourteen new real-PostgreSQL lifecycle cases and the existing startup guard pass. +The producer-removal regression fails as expected when fast-response settlement +has no notification. Migration application/replay pass after adding the generated +concurrent-index migration to the existing top-level execution registry. Broader +local and remote acceptance is pending. See the detailed +[GitHub owner audit](2026-10-25-github-chat-maintenance-sleep-audit.md). + +All earlier failing CI, review findings, stack/master reconciliation, and remaining +inventory duties retain their existing requirements. This slice does not claim +that the stack is ready or authorize a merge. + +Slice-34 final focused acceptance passes 17 new cases plus the startup guard. +Workspace typecheck/build pass again after the receipt-lifecycle correction. +The fast-response suites pass 69 tests. The initial complete chat run passed +1,554 tests; a fresh complete run with the three added lifecycle regressions +is in progress. No remote gate or full-stack readiness claim is made yet. diff --git a/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md b/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md new file mode 100644 index 00000000000..4d340c9ebab --- /dev/null +++ b/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md @@ -0,0 +1,409 @@ +# Remaining GitHub chat maintenance owners + +Inspected after slice 33 at `6d01f4c730aed91e660a63eede688f4f4da61252`. +No implementation or verification claim yet. + +The label "GitHub history maintenance" hides several separate duties. +`chat-channels/github/maintenance.ts` serially calls response-comment +`processPending`, then review-check `processPending`. It launches through the +chat service's tracked message-processing lifecycle, with a single-flight flag. +Do not replace that tracked work with an unjoined promise or shared-provider code. + +## Candidate slice 34: response-comment acknowledgement dispatch + +Owner: `chat-github-response-comments.ts`. Existing durable actions have +kind `github_response_comment`. `stageResponse` inserts status received before +competing for the repository publication lease, so contention does not lose work. +It is called both by acknowledgement and `writeGitHubResponseComment`, including +agent-authored/final response paths. Insertion currently uses a root client; +notification must be in a transaction around the actual write. Repeated calls +also read the existing action, and must not repeatedly wake themselves solely +because an insert lost onConflictDoNothing. + +The first half of `processPending` selects received actions and failed actions +with result.retryable=true whose saved retryAt is due, oldest-created first, +limit 10. `tryAcknowledge` records attempts and exponential retryAt capped at +five minutes; existing policy stops automatic retries after eight attempts. +403 authorization changes cancel the action. Preserve that current policy, +company/endpoint/principal authority, generation fencing, and lease ownership. +Do not infer successful delivery from a missing row or an SDK return alone. + +Acknowledgement can deliberately defer to a fast response while its request or +publication is pending/running/streaming, or until its saved expiresAt. Published +or ambiguous fast responses supersede the ordinary acknowledgement without +another send. A worker cannot simply repeatedly run every received action with +now+backoff: park behind committed fast-response/publication state and retain +only the real expiration/recovery obligation. Inspect all those producers. + +Provider history scanning is bounded at 100 pages of 100 comments and fails +closed if identity cannot safely be recovered. Preserve App-owned markers, +shared pull-response targets, legacy placeholder retirement, no-regression final +receipts, and uncertain-post handling. Provider details stay in this owner. + +## Separate reason: terminal response reconciliation + +The second half of response `processPending` scans processed actions with +result.final != true. It joins conversation/endpoint/issue and latest terminal +run for the assigned agent, with blocked-issue or source-message provenance +conditions. Acknowledgement completion is not final-response completion. +Run transitions, issue changes, source linkage, response/publication settlement, +and fast-response changes can affect eligibility. Current milestone notifications +may provide an existing boundary but require a full writer audit. + +`acknowledge` preserves source-run/active-identity checks and defers while +`githubRunReplyState` is unsettled. A successful final reply may require retiring +an older placeholder rather than posting a second reply. Existing no-op maintenance +writes updatedAt only to rotate oldest-prefix candidates; removing that polling +must replace starvation handling without manufacturing permanent writes. +Waiting for a nonterminal run is passive, while unresolved actual delivery still +needs owned recovery. Keep this independently reviewable from initial dispatch. + +## Separate reason: GitHub review-check publication + +Owner: `chat-github-checks.ts`, kind `github_review_check`. Enqueue comes from +webhook-events and chat-github-reviews. It snapshots runtime generation and +normalized review event. Delivery derives queued/running/completed state from +review, run, inbound delivery, assessment and current configuration. Existing +repository publication leases and GitHub head/authority checks remain mandatory. +A terminal assessment has its own outbox; ordinary check delivery marks itself +processed when that assessment owns the result. Never dispatch agents here. + +Nonterminal success currently saves received plus retryAt=now+5s, even if +nothing changed. Replace that state discovery with producer events. Failed +publication has saved exponential retryAt and the existing eight-attempt policy. +Investigate stale-head discovery: an external PR push must have a supported +webhook/replay contract or a scoped outstanding-check reconciliation deadline. +Do not remove its only fallback on assumption that local run events cover it. + +## Separate reason: late review projection repair + +Before dispatch, check processPending globally updates queued/running review +rows without assessments when a matching cancelled stale-head check receipt +exists. This repairs admission that commits after cancellation. Both writers +must signal it; late review insert and cancellation are separate commits. +Extract this duty from publication dispatch rather than keep a global update +on every check wake. Preserve exact company/endpoint/delivery/event identity. + +All four duties still need real producer tests, startup reconstruction, scoped +fair progress, rollback/commit, drain/resume, and quiet retained-history tests. +The aggregate maintenance call cannot be retired until every duty is covered. + +## Additional producer/dependency evidence + +Response writes also enter from `chat-github-reviews.ts` (agent comment/update +and assessment paths), `chat-channels/publications/transport.ts` (failed-run +publication), and `github/wake-authority.ts` after inbound admission. The +`tryAcknowledge` boundary classifies current authority failures as cancellation; +it does not silently return null for a missing source. Preserve that distinction. + +Fast-response insertion already signals `DELIVERY_QUEUES.fastResponse` through +`fast-responses.ts`, but that notification currently has no scope and later +request-state transitions need their own audit. Publication notifications carry +optional issue scope only on relevant writers; not every publication mutation +currently fans out to milestones. A future acknowledgement worker must cover +both request and publication transitions without importing GitHub policy into +these shared services. The concrete cross-owner capability is notification of +a change to an existing response slot, not a GitHub-specific scheduler branch. + +Existing integration fixtures and regression tests are in the early GitHub +section of `chat-channels.integration.test.ts` (around lines 2,750-6,200 in this +revision); there are no standalone response-comment or check suites at the +initial guessed filenames. Reuse the controlled provider API and real DB +fixtures rather than replace authority/lease operations with mocks. + +## Design constraints refined during slice 33 acceptance + +Keep initial acknowledgement dispatch separate from terminal-response recovery. +The public/manual `processPending` can delegate to both, while recurring +maintenance calls only the terminal half until its own later migration. + +The repository publication lease already releases through a transaction and +signals `chatGitHubPublications`. A new acknowledgement worker must not register +a second worker under that occupied topic. It needs its own topic and narrowly +scoped lease-release hints or a shared response-availability notification. +Lost/contended lease acquisition returns undefined; that is not proof of a +successful acknowledgement. Represent the actual outstanding lease expiry and +react to release. Renewals are active-operation work, not idle polling. + +Avoid a self-wake loop: `stageResponse` is called repeatedly while acknowledging +an existing action. A no-op insert/read must not repeatedly trigger dispatch of +that same received action. A received action waiting on fast-response publication +needs either the saved fast-response expiry, publication outcome changes, or +concrete unresolved claim recovery. Never return now indefinitely for it. + +Most ordinary publication terminal writers already supply company/issue scope to +`notifyChatPublicationWork`; provider retry/unknown and stale-stream quarantine +paths sometimes omit it. Specifically inspect `publications/failures.ts`, +`worker.ts::settlePublicationPreparationFailure`, and queue quarantine. A fast +response that changes streaming -> retry/unknown must release/reclassify the +acknowledgement's wait. Do not depend solely on succeeded publication events. + +Fast-response state writers include enqueue, skip, settle, claim, publication +comment binding, supersede, final publication-status suppression, and bulk expiry. +Only changes to the actual acknowledgement predicate require a wake. Expiry is +already a saved deadline; claim pending -> running keeps the same predicate. +Early failure/skip and adding a publication binding require producer coverage. +Do not blindly wrap every write or introduce GitHub branches in fast-responses. +Use a normalized response-slot change contract if these shared owners need a +new notification capability, with existing transactions reused wherever present. + +## Slice 34 implementation and boundaries + +Initial response acknowledgement now registers with the existing delivery +coordinator. This is a separate worker from terminal-response reconciliation, +review-check delivery, and stale-head repair. The latter three remain in the +GitHub maintenance owner. The explicit/manual combined `processPending` entry +point still performs both initial and terminal response recovery. + +The durable source remains `chat_actions.kind = github_response_comment`. +Received actions and retryable failed actions are reconstructed on startup in +100-row ID pages. The shared scoped-deadline helper dispatches a fair batch of +25 across companies. Exact delivery, conversation, issue, and repository-lease +hints restrict reconstruction. Processed/cancelled actions and exhausted failures +leave this worker's inventory; retained response history does not arm a timer. +No new queue table or changed automatic retry limit is introduced. + +The next deadline is derived from existing saved retry state, a competing +repository publication lease, or a pending fast response's expiration. A +streaming fast response parks acknowledgement work until publication settlement +or quarantine changes its state. The publication worker still owns stale-stream +recovery. This worker does not invent a second timeout or replay an uncertain +publication. Published or delivery-unknown fast responses suppress the ordinary +acknowledgement through the existing provider owner. + +Producer coverage: + +- `stageResponse` signals its insert transaction. Existing response lookups do + not emit another signal. This prevents acknowledgement itself from continuously + waking an unchanged received action. +- Failed/cancelled acknowledgement settlement signals the existing action's + delivery before its transaction writes the result. +- Repository publication lease release signals the precise endpoint/repository/ + pull-or-issue number. The signal covers receipt changes made under that lease, + and releases waiting acknowledgements when another publication finishes. +- Fast-response admission, early skip, accounting settlement, and publication + binding signal the existing response slot. Pending-to-running claims retain + the same wait predicate; pending expiry already has a saved deadline. +- `supersedeFastResponse` is used only by Board chat, which has no external + acknowledgement delivery. It was audited and left unchanged. +- Publication completion, failure, retry, ambiguous delivery, preparation failure, + pause-race reset, and stale-stream quarantine notify response availability. + Task-control cancellation covers the affected conversation. Content preparation + and lease bookkeeping do not create response-availability debt. +- Endpoint removal deletes its response actions. Other authority checks remain + in the response service before any provider write. The worker does not cache + authorization or infer authority from a queued action. + +The shared capability is `notifyChatResponseAvailability`: an identifier-only +hint that an existing response slot may have changed. Fast-response and ordinary +publication owners do not inspect GitHub payloads or choose GitHub behavior. +Only the GitHub acknowledgement owner parses repository thread identity and +applies its response protocol. Notification serialization explicitly picks IDs +so passing a publication row cannot carry its payload into scheduler hints. + +Provider requests remain outside database transactions. The existing repository +lease, current source/principal checks, response markers, shared pull-response +identity, version fencing, and final-receipt preservation remain in charge of +provider effects. A recovered known receipt is marked processed so it cannot +remain permanently in the pending acknowledgement inventory. Admission is checked +at the actual fetch boundary after lease renewal; an in-flight request may settle, +but a stopped worker cannot start the next provider request. Stop-related failures +do not consume the provider retry budget. + +The migration adds three concurrent, idempotent indexes: pending acknowledgement +IDs, fast-response delivery lookup, and publication comment/destination lookup. +It uses the existing top-level migration mechanism and snapshot pruning rules. +These indexes avoid rebuilding pending work by scanning completed history. + +Supported writes are the service's process-local transaction notification model. +Independent SQL writers or a second instance do not acquire a new wake guarantee +from this change. Host sleep eligibility and external alarm coverage are unchanged. +A local deadline is not evidence that a stopped host can wake itself. + +## Verification evidence retained during development + +- The first eight lifecycle tests failed on a PostgreSQL JSON-expression precedence + error in the lease lookup. Parenthesizing the repository-ID expression fixed + the query; all eight then passed. +- Removing the real fast-response settlement notification made the regression + fail: fallback waited until the response expired. Restoring that signal is the + implementation under test. The assertion measures when the action settles, + separately from the coordinator's final coalesced bookkeeping pass. +- The initial concurrent-index application failed because the migration was not + in the existing top-level statement registry. Registering it fixed application; + a second application also completed successfully. +- Expanded tests exposed provider error normalization at the fetch boundary. + The worker now preserves a stopped attempt without charging a provider failure. + The fault fixture was also narrowed to response-comment POSTs so installation + access-token requests were not mistaken for acknowledgement attempts. +- Fourteen new real-PostgreSQL cases and the startup guard pass. They cover empty + and retained-history quiet, lease release, restart, drain, standby, fast-response + failure/expiry/publication/unknown outcome, stale-stream quarantine, outer + rollback/commit, saved provider retry, and stop between provider requests. +- Full chat/fast-response acceptance, workspace typecheck/build, remote CI and + review remain pending. This document does not yet claim PR readiness. + +Workspace typecheck/build now pass. The fast-response test harness supports an +explicit fresh `_test` PostgreSQL database, retaining embedded PostgreSQL by +default. Its first run passed the 69 test bodies but failed cleanup on an invalid +client-close call; the corrected setup then hit retained fixed-prefix fixtures +when the same database was reused. A fresh database passes all 69 tests and cleanup. +Final server typecheck passes. All three migration indexes are valid in PostgreSQL. +The complete affected chat suites are still running; remote checks/review have not +been claimed complete. + +A rollback-only PostgreSQL plan experiment seeded 10,000 completed response +actions and 10,000 unrelated fast-response records alongside one pending action +and its matching request. The pending-ID lookup used +`chat_actions_github_ack_work_idx` and read one row (two shared buffers); the +request lookup used `fast_response_requests_delivery_idx` and read one row +(three shared buffers). This checks the two lookup access paths, not whole-worker +latency or deployed fleet savings. All experimental writes were rolled back. + +The first complete chat acceptance passed all 1,554 tests without skips in +496.42 seconds. Final review then added three receipt-lifecycle cases: confirmed +received/failed receipts settle, but cancelled receipts stay cancelled. The new +cancelled case failed before correction. Settlement now applies only to pending +states and compares the observed status before writing. The final version of the +producer-removal test also failed as expected without the settlement notification. +The corrected implementation is undergoing a fresh full acceptance run with the +three additional cases; earlier full results are not substituted for that run. + +After the final receipt correction, all 17 new lifecycle cases and the startup +guard pass. Workspace typecheck and build pass again. The final complete chat +run remains in progress; source and branch remain fixed during acceptance. + + +## Acknowledgement presentation wait correction + +An acknowledgement for the latest successful native run can remain `received` +while that run's presentation decision is still absent. Reattempting it only +acquires/releases the publication lease and returns, leaving an overdue entry. +Recovery now shares the delivery owner's request/run projection and parks that +specific no-reply state. It retains normal lease, fast-response, and delivery +retry deadlines for actionable requests. Run/presentation/publication changes +wake it through the existing transaction-aware response-availability topic. +No new rows or periodic recovery scan are introduced. + +The GitHub owner keeps request binding and provider reply interpretation in +`github/response-run-state.ts`; both delivery and scheduling use the same +projection. Heartbeat's existing presentation write is named +`recordRunPresentationDecision` without changing its transaction or event order, +so the real producer can be exercised with outer commit and rollback. This does +not move GitHub publication authority into the generic native chat repair path. + +The first usable regression reproduced an overdue deadline in +`/tmp/sleep-ack-presentation-red-terminal.log`. Earlier fixture attempts lacked +GitHub configuration or terminal success. An attempted generic native-response +repair correctly returned false: GitHub replies remain tool-owned. Those fixture +failures remain in the red/configured, green, and current logs. The final regression +uses the existing GitHub delivery fixture and real heartbeat decision writer: +no SQL for two virtual minutes while parked or after rollback; commit updates the +same provider comment once, then returns to quiet. + +All 228 GitHub workflow cases pass (`/tmp/sleep-ack-presentation-acceptance.log`), +as do all 30 heartbeat completion cases (`/tmp/sleep-ack-completion-full.log`) and +server typecheck (`/tmp/sleep-ack-presentation-types-final.log`). The initial +completion command omitted its database URL and skipped 29 cases; the full result +uses a fresh dedicated migrated database. No full-repository, final-stack, or +deployed sleep claim follows from these owner-level checks. + + +## Current parent and migration acceptance — October 28 + +The existing GitHub acknowledgement slice is normally integrated onto published +continuation parent `a1d16f84b3fae361d5d648670a5964486370c3a3`. Integration +`51f2f35f54d75648e50f9ec9beada1e479307ecb` preserves every original domain/schema +edit and every unrelated parent blob. It carries the parent confirmation and +continuation work indexes. Proof: +`/tmp/sleep34-indexed-parent-composition-proof.json`. + +The old draft-owned migration 0394 collided with landed history. Its three +indexes were first regenerated locally as 0399; the preceding continuation then +needed its own proven work index at 0399. This slice now generates +`0400_absent_phantom_reporter.sql` against the current schema. Only unlanded +artifacts were replaced. The three original concurrent/idempotent SQL index +definitions are identical after whitespace normalization. All prior SQL files, +journal entries and retained snapshot bytes remain intact. Canonical generation +prunes 0395 and keeps the five newest snapshots, 0396–0400. The top-level +migration registry includes the new filename. Both normal merge parents and the +exact tree are checked; no force push or rebase occurs. + +At `337f272853d4f8a51b5b9300900a46eb4e13f698`, **305 distinct cases pass**: + +- All 236 selected GitHub agent-review workflow cases, with 1,485 other chat + cases skipped. This includes acknowledgement lease/restart/drain/standby, + fast-response outcomes, outer rollback/commit, saved retry, stop between sends, + receipt-state preservation and the real heartbeat presentation writer. +- All 26 cases in the complete fast-response service suite. +- All 30 cases in the complete heartbeat completion suite, with an explicit + external database so no embedded-platform guard skips the cases. +- All 13 app/startup cases. + +Database build, server typechecking/build, module boundaries, three fresh +PostgreSQL migrations through 0400 and the exact parent-to-head migration-order +gate pass. The first migration-order invocation passed symbolic `HEAD` to a +command that requires a full 40-character SHA; its usage output is retained. +The corrected command passes without a code change. + +An existing populated 0399 test database also upgrades to 0400: all 398 prior +migration receipts remain identical, exactly one receipt is added, and replay +adds none. All three indexes are valid and ready in both a fresh and upgraded +database. Receipt: `/tmp/sleep34-upgrade-replay-receipt.json`. This tests schema +application and replay; it does not claim whole-worker latency or fleet savings. + +Commands and exact fixtures: `/tmp/sleep34-current-acceptance-command.json`. +Logs/exit receipts: `/tmp/sleep34-current-{db-build,types,build,boundaries,github,fast,completion,startup}.*` +and `/tmp/sleep34-current-migrate-{github,fast,completion}.*`. The earlier 69-case +fast-response group, 228-case GitHub group and broad chat runs above retain their +original revision scopes. In particular, the current complete service file has +26 cases; an earlier broader count is not substituted for it. + +This remains local preparation while #17653 finishes current-head remote gates. +Its eventual landed squash parent must be reconciled before publication. The +following acceptance-only edit leaves all owned source/test/migration hashes +unchanged. Fresh exact-head CI and Greptile 5/5 with no actionable findings are +still required for this slice. No full local repository suite, live GitHub tenant +or deployed sleep result is inferred from this acceptance. + + +## Acceptance on the actual landed continuation parent + +Continuation recovery #17653 has merged at +`063cfd43813cabf18417605fb9eefb9aacf95c8d`. Normal two-parent integration +`9b3607e159153e00ccdfe538d52b591bd38d6499` preserves the full 28-path owned +binary patch and all 20 intervening master paths. Those paths include native +retry recovery, OpenCode provider-event exit codes and evaluation report-lock +workflow fixes. The verified published parent is the semantic base for historic +squash conflicts. Both normal parents and the exact tree are checked; there is +no force push or rebase. Proof: +`/tmp/sleep34-landed-parent-publication-proof.json`. + +**87 distinct current cases pass on that landed-parent source:** all 18 selected +acknowledgement lifecycle/presentation cases (1,703 other chat cases skipped), +all 26 fast-response service cases, all 30 heartbeat completion cases and all 13 +startup cases. Database build, server typechecking/build, module boundaries, +three fresh PostgreSQL migrations through 0400 and the current-master migration +order gate pass. The Runner vendor bundle is rebuilt successfully because its +upstream provider-event inputs changed. Database build runs first so the prior +local terminal-response preparation's ignored 0401 output cannot leak into this +0400 acceptance. + +This selection rechecks the supported signals, claims, rollback/commit, +presentation parking, stop boundaries and settlement consumers after the master +context changes. The earlier 305-case matrix remains scoped to prepared revision +`337f272853d4f8a51b5b9300900a46eb4e13f698`; the repeated 87 cases are not added to +305. All three index definitions and schema history are unchanged during this +integration. The populated upgrade/replay and fresh/upgrade index-validity proof +above therefore keeps its stated schema scope. + +Commands, revision and actual fixtures: +`/tmp/sleep34-landed-acceptance-command.json`. Receipts/logs: +`/tmp/sleep34-landed-{db-build,types,build,boundaries,github,fast,completion,startup}.*`, +`/tmp/sleep34-landed-migrate-{github,fast,completion}.*`, and +`/tmp/sleep34-landed-migration-order.*`. +The following audit-only commit leaves all owned source/test/migration hashes +unchanged, recorded in `/tmp/sleep34-landed-tested-source-sha256.json`. +Original failures and earlier scopes above remain visible. Publication requires +fresh exact-head CI and Greptile 5/5 with no actionable findings. No full local +repository, live GitHub tenant or deployed sleep result is inferred. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 22b07ed7279..859c3f52bc3 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -551,6 +551,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0397_known_madrox.sql", "0398_yielding_random.sql", "0399_ambitious_onslaught.sql", + "0400_absent_phantom_reporter.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0400_absent_phantom_reporter.sql b/packages/db/src/migrations/0400_absent_phantom_reporter.sql new file mode 100644 index 00000000000..ca4eafbda46 --- /dev/null +++ b/packages/db/src/migrations/0400_absent_phantom_reporter.sql @@ -0,0 +1,5 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_ack_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_response_comment' + and "chat_actions"."delivery_id" is not null and coalesce("chat_actions"."result"->>'final', 'false') <> 'true' + and ("chat_actions"."status" = 'received' or ("chat_actions"."status" = 'failed' and "chat_actions"."result"->>'retryable' = 'true'));--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_publications_comment_destination_idx" ON "chat_publications" USING btree ("company_id","comment_id","endpoint_id","conversation_id") WHERE "chat_publications"."comment_id" is not null;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "fast_response_requests_delivery_idx" ON "fast_response_requests" USING btree ("company_id","endpoint_id","delivery_id") WHERE "fast_response_requests"."delivery_id" is not null; diff --git a/packages/db/src/migrations/meta/0395_snapshot.json b/packages/db/src/migrations/meta/0400_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0395_snapshot.json rename to packages/db/src/migrations/meta/0400_snapshot.json index 02a4b17b992..968429e6801 100644 --- a/packages/db/src/migrations/meta/0395_snapshot.json +++ b/packages/db/src/migrations/meta/0400_snapshot.json @@ -1,6 +1,6 @@ { - "id": "66fe02b3-6672-4c19-b447-57acad43bf16", - "prevId": "633afb6b-4b89-44ea-81d5-a477a3696eb5", + "id": "843ab3c7-1378-42f3-b213-401a8d303e3f", + "prevId": "9ca01952-f1fa-4c54-a189-3f5a3a3a3761", "version": "7", "dialect": "postgresql", "tables": { @@ -9624,6 +9624,50 @@ } }, "indexes": { + "chat_actions_continuation_work_idx": { + "name": "chat_actions_continuation_work_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'interaction_wakeup' AND \"chat_actions\".\"status\" IN ('issued', 'processing')", + "concurrently": false, + "method": "btree", + "with": {} + }, + "chat_actions_confirmation_recovery_idx": { + "name": "chat_actions_confirmation_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'confirmation_response' AND \"chat_actions\".\"status\" = 'processing'", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_actions_provider_action_uq": { "name": "chat_actions_provider_action_uq", "columns": [ @@ -9644,6 +9688,22 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_actions_github_ack_work_idx": { + "name": "chat_actions_github_ack_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"delivery_id\" is not null and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -12071,6 +12131,40 @@ "method": "btree", "with": {} }, + "chat_publications_comment_destination_idx": { + "name": "chat_publications_comment_destination_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "comment_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "conversation_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_publications\".\"comment_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_publications_idempotency_uq": { "name": "chat_publications_idempotency_uq", "columns": [ @@ -14188,6 +14282,24 @@ "primaryKey": false, "notNull": false }, + "provider_account_id": { + "name": "provider_account_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "cleanup_confirmed_at": { + "name": "cleanup_confirmed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "cleanup_next_attempt_at": { + "name": "cleanup_next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, "expires_at": { "name": "expires_at", "type": "timestamp with time zone", @@ -14210,6 +14322,22 @@ } }, "indexes": { + "chat_voice_inbound_cleanup_idx": { + "name": "chat_voice_inbound_cleanup_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_voice_inbound_expiry_idx": { "name": "chat_voice_inbound_expiry_idx", "columns": [ @@ -14962,6 +15090,28 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_voice_sessions_report_window_idx": { + "name": "chat_voice_sessions_report_window_idx", + "columns": [ + { + "expression": "ended_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -30460,6 +30610,34 @@ "method": "btree", "with": {} }, + "fast_response_requests_delivery_idx": { + "name": "fast_response_requests_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "delivery_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"fast_response_requests\".\"delivery_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "fast_response_requests_history_idx": { "name": "fast_response_requests_history_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 62fad4474c4..a434b53ace1 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2787,6 +2787,13 @@ "when": 1793218102114, "tag": "0399_ambitious_onslaught", "breakpoints": true + }, + { + "idx": 400, + "version": "7", + "when": 1793218595176, + "tag": "0400_absent_phantom_reporter", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/chat_channels.ts b/packages/db/src/schema/chat_channels.ts index 1bb91f2603c..0e0f56abf4b 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -564,6 +564,8 @@ export const chatPublications = pgTable( name: "chat_publications_company_comment_fk", }), index("chat_publications_work_idx").on(table.state, table.nextAttemptAt), + index("chat_publications_comment_destination_idx").on(table.companyId, table.commentId, table.endpointId, table.conversationId) + .where(sql`${table.commentId} is not null`), uniqueIndex("chat_publications_idempotency_uq").on( table.companyId, table.idempotencyKey, @@ -683,6 +685,9 @@ export const chatActions = pgTable( table.endpointId, table.providerActionId, ), + index("chat_actions_github_ack_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment' + and ${table.deliveryId} is not null and coalesce(${table.result}->>'final', 'false') <> 'true' + and (${table.status} = 'received' or (${table.status} = 'failed' and ${table.result}->>'retryable' = 'true'))`), foreignKey({ columns: [table.companyId, table.deliveryId], foreignColumns: [chatDeliveries.companyId, chatDeliveries.id], diff --git a/packages/db/src/schema/fast_responses.ts b/packages/db/src/schema/fast_responses.ts index 2dfdb72ef17..a7fb859d33b 100644 --- a/packages/db/src/schema/fast_responses.ts +++ b/packages/db/src/schema/fast_responses.ts @@ -81,6 +81,8 @@ export const fastResponseRequests = pgTable( t.companyId, t.sourceKey, ), + delivery: index("fast_response_requests_delivery_idx").on(t.companyId, t.endpointId, t.deliveryId) + .where(sql`${t.deliveryId} is not null`), history: index("fast_response_requests_history_idx").on( t.companyId, t.acceptedAt, diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 7123ae6c78e..c282a0dfd7f 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -94,6 +94,10 @@ describe("createChatReconciliationCoordinator", () => { expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications("); expect(publicationMaintenance).not.toContain("reconcileTerminalConfirmationActions("); expect(source).toContain("chatChannels.registerInteractionContinuationWork(deliveryWork,"); + expect(source).toContain("chatChannels.registerGitHubAcknowledgementWork(deliveryWork,"); + const github = readFileSync(new URL("../services/chat-channels/github/maintenance.ts", import.meta.url), "utf8"); + expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processPending()"); + expect(github).toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()"); expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,"); expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,"); }); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index b84bda2b3b5..7ec86d1574b 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,4 @@ +import { recordRunPresentationDecision } from "../services/heartbeat/run-completion.js"; 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"; @@ -43,7 +44,7 @@ import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-wor import { aiConnectionService } from "../services/ai-connections.js"; import { fastResponseService } from "../services/fast-responses.js"; import { fastResponseReceipt } from "../services/fast-response-provider.js"; -import { fastResponseRequests } from "@paperclipai/db"; +import { companyFastResponses, fastResponseRequests } from "@paperclipai/db"; import { toolAccessService } from "../services/tool-access.js"; import { toolActionRequests, toolInvocations, toolPolicies } from "@paperclipai/db"; import { GitHubPublicationLeaseLost, withGitHubPublicationLease } from "../services/chat-github-publication-lease.js"; @@ -4035,6 +4036,240 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(attempts).toBe(2); expect(f.responseWrites).toHaveLength(writes); }); + it.each(["release", "restart", "drain", "standby"] as const)("wakes GitHub acknowledgements from a real publication lease %s", async mode => { + const f = await reviewBotFixture(); + const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id)); + const errors: unknown[] = []; + let canRun = true; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + const register = () => f.service.registerGitHubAcknowledgementWork(coordinator, () => canRun); + let worker = register(); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const claimed = new Promise(resolve => { entered = resolve; }); + let lease: Promise | undefined; + try { + await worker.ready; + expect(errors).toEqual([]); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + expect(coordinator.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + vi.setSystemTime(Date.now() - 120_000); + lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id, + repositoryId: String(resource.metadata.providerRepositoryId), number: 418 }, f.providerFetch, async () => { entered(); await hold; }); + await claimed; + if (mode === "restart") await coordinator.stop(); + if (mode === "drain" || mode === "standby") canRun = false; + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("received"); expect(f.responseWrites).toEqual([]); + release(); await lease; + if (mode === "drain" || mode === "standby") { + await vi.advanceTimersByTimeAsync(1000); + expect(f.responseWrites).toEqual([]); + canRun = true; worker.wake(); + } + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = register(); await worker.ready; + } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].status).toBe("processed"); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(f.responseWrites).toHaveLength(1); + expect(f.responseWrites[0].body).toContain("Working on this"); + const settledReads = vi.spyOn(db, "select"), settledWrites = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(settledReads).not.toHaveBeenCalled(); expect(settledWrites).not.toHaveBeenCalled(); + settledReads.mockRestore(); settledWrites.mockRestore(); + expect(errors).toEqual([]); + } finally { + release(); await lease; await coordinator.stop(); vi.useRealTimers(); + } + }); + + it.each(["failure", "expiry", "published", "delivery_unknown", "streaming quarantine"] as const)("releases a GitHub acknowledgement after fast response %s without maintenance polling", async outcome => { + await instanceSettingsService(db).updateExperimental({ enableFastResponses: true }); + const f = await reviewBotFixture(); + const binding = await aiConnectionService(db).save(f.companyId, "owner-user", { provider: "openrouter", method: "api_key", name: "Fast GitHub", ownership: "shared", apiKey: "fixture-key", agentIds: [], allAgents: true }, "fixture-key"); + const fast = fastResponseService(db, { authorizeExternal: f.service.authorizeFastResponse, + provider: async () => ({ ...(outcome === "failure" ? { errorCode: "invalid_output", noProviderWork: true } : { text: "I’ll check it." }), receipt: fastResponseReceipt({ usage: { inputTokens: 80, outputTokens: 8 } }) }) }); + await fast.configure(f.companyId, "owner-user", { enabled: true, ...binding, model: "openai/gpt-oss-120b", allowSponsored: true }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [job] = await db.select().from(fastResponseRequests).where(eq(fastResponseRequests.companyId, f.companyId)); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime()); + }, { timeout: 5000 }); + expect(f.responseWrites).toEqual([]); + if (outcome === "expiry") await vi.advanceTimersByTimeAsync(job.expiresAt.getTime() - Date.now()); + else { + await fast.process(job); + if (outcome === "streaming quarantine") { + await db.update(chatPublications).set({ state: "streaming", updatedAt: new Date(0) }) + .where(eq(chatPublications.idempotencyKey, `fast-response:${job.id}`)); + await vi.advanceTimersByTimeAsync(Math.max(0, job.expiresAt.getTime() - Date.now())); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + expect(f.responseWrites).toEqual([]); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + await f.service.processPendingPublications(); + } else if (outcome === "delivery_unknown") { + await db.transaction(async tx => { + await notifyChatPublicationWork(tx, { companyId: job.companyId, issueId: job.issueId }); + await tx.update(chatPublications).set({ state: "delivery_unknown" }).where(eq(chatPublications.idempotencyKey, `fast-response:${job.id}`)); + }); + } else if (outcome === "published") await f.service.processPendingPublications(); + } + let processedAt: number | undefined; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("processed"); + processedAt ??= Date.now(); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + if (outcome !== "expiry" && outcome !== "streaming quarantine") expect(processedAt).toBeLessThan(job.expiresAt.getTime()); + expect(f.responseWrites).toHaveLength(["failure", "expiry"].includes(outcome) ? 1 : 0); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("wakes a GitHub acknowledgement only after the outer fast-response transaction commits", async () => { + await instanceSettingsService(db).updateExperimental({ enableFastResponses: true }); + const f = await reviewBotFixture(); + const binding = await aiConnectionService(db).save(f.companyId, "owner-user", { provider: "openrouter", method: "api_key", name: "Fast GitHub", ownership: "shared", apiKey: "fixture-key", agentIds: [], allAgents: true }, "fixture-key"); + const fast = fastResponseService(db, { authorizeExternal: f.service.authorizeFastResponse }); + await fast.configure(f.companyId, "owner-user", { enabled: true, ...binding, model: "openai/gpt-oss-120b", allowSponsored: true }); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [job] = await db.update(fastResponseRequests).set({ expiresAt: new Date(Date.now() + 60_000) }) + .where(eq(fastResponseRequests.companyId, f.companyId)).returning(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime()); + const skip = async (tx: Parameters[0]>[0]) => { + await tx.update(companyFastResponses).set({ enabled: false }).where(eq(companyFastResponses.companyId, f.companyId)); + expect(await fastResponseService(tx as unknown as Db).process(job)).toMatchObject({ status: "unavailable" }); + await vi.advanceTimersByTimeAsync(100); + expect(f.responseWrites).toEqual([]); + expect((await db.select().from(fastResponseRequests).where(eq(fastResponseRequests.id, job.id)))[0].status).toBe("pending"); + }; + await expect(db.transaction(async tx => { await skip(tx); throw new Error("rollback fast response skip"); })).rejects.toThrow("rollback fast response skip"); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(idleWorkSnapshot().active).toBe(0); }); + expect(f.responseWrites).toEqual([]); + expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime()); + await db.transaction(skip); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect(f.responseWrites).toHaveLength(1); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 10_000 }); + expect(Date.now()).toBeLessThan(job.expiresAt.getTime()); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["drain", "standby", "shutdown"] as const)("stops GitHub acknowledgement sends at %s between provider requests", async mode => { + const f = await reviewBotFixture(); + f.setSupplementalProviderFetch(async (input, init) => String(input).includes("/issues/418/comments") && init?.method === "POST" ? new Response("unavailable", { status: 503 }) : undefined); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("failed"); expect(action.result?.attempts).toBe(1); + await db.update(chatActions).set({ result: { ...action.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, action.id)); + let release!: () => void, entered!: () => void; + const held = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + f.setSupplementalProviderFetch(async input => { + if (String(input).includes("/issues/418/comments?")) { entered(); await held; } + return undefined; + }); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + let worker = f.service.registerGitHubAcknowledgementWork(coordinator, () => canRun); + let stopping: Promise | undefined; + try { + await started; + if (mode === "shutdown") stopping = coordinator.stop(); else canRun = false; + release(); await worker.ready; await stopping; + expect(f.responseWrites).toEqual([]); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].result?.attempts).toBe(1); + canRun = true; + if (mode === "shutdown") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = f.service.registerGitHubAcknowledgementWork(coordinator, () => true); await worker.ready; + } else worker.wake(); + await vi.waitFor(() => { expect(f.responseWrites).toHaveLength(1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { release(); await stopping; await coordinator.stop(); } + }); + + it("retries a failed GitHub acknowledgement at its saved deadline", async () => { + const f = await reviewBotFixture(); + let unavailable = true, attempts = 0; + f.setSupplementalProviderFetch(async (input, init) => { + if (String(input).includes("/issues/418/comments") && init?.method === "POST") { attempts++; if (unavailable) return new Response("unavailable", { status: 503 }); } + return undefined; + }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) }); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.status).toBe("failed"); expect(attempts).toBe(1); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(coordinator.nextWakeAt()).toBe(Date.parse(String(action.result?.retryAt))); }); + unavailable = false; + await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now())); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(f.responseWrites).toHaveLength(1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 }); + expect(attempts).toBe(2); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["received", "failed", "cancelled"] as const)("preserves the %s GitHub receipt lifecycle during acknowledgement recovery", async status => { + const f = await toolReplyFixture(); + const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + expect(action.result?.id).toBeTruthy(); + await db.update(chatActions).set({ status }).where(eq(chatActions.id, action.id)); + const writes = f.responseWrites.length; + await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(action.deliveryId!); + const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, action.id)); + expect(settled.status).toBe(status === "cancelled" ? "cancelled" : "processed"); + expect(settled.result).toEqual(action.result); + expect(f.responseWrites).toHaveLength(writes); + }); + it("does not post a GitHub working comment when the scheduler declines the request", async () => { const f = await reviewBotFixture(); f.wakeup.mockImplementation(async (agentId, opts) => { @@ -4167,6 +4402,54 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect([...f.responseComments.values()][0].body).toContain("paperclip-response-state:final"); expect(f.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]); }); + it("parks a GitHub acknowledgement until native presentation commits", async () => { + const f = await toolReplyFixture(); + const [action] = await db.select().from(chatActions).where(and( + eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + await db.update(chatActions).set({ status: "received", result: null }).where(eq(chatActions.id, action.id)); + await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "committed", resultJson: {} }) + .where(eq(heartbeatRuns.id, f.run.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => errors.push(error) }); + const decision = { schema: "paperclip.run_presentation_decision.v1" as const, + commentAction: "none" as const, chosenSource: "none" as const, reasonCodes: ["no_final_response"] }; + const commit = (database: typeof db) => recordRunPresentationDecision(database, { + companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, decision, + }); + try { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready; + const quiet = async () => { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); + expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + }; + await quiet(); + expect(f.responseWrites).toHaveLength(1); + await expect(db.transaction(async tx => { + await commit(tx as unknown as typeof db); + expect((await readHeartbeatRunsById(f.run.id))[0].resultJson?.presentationDecision).toBeUndefined(); + throw new Error("rollback presentation"); + })).rejects.toThrow("rollback presentation"); + await quiet(); + expect(f.responseWrites).toHaveLength(1); + await commit(db); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(10); + expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0]) + .toMatchObject({ status: "processed", result: { final: true } }); + }, { timeout: 10_000 }); + expect(f.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]); + expect(f.responseWrites[1].body).toContain("without publishing a final response"); + await quiet(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); it("does not let an unsettled GitHub fallback block a later board message", async () => { const f = await toolReplyFixture(); await db.insert(chatActions).values({ companyId: f.companyId, endpointId: f.endpoint.id, conversationId: f.conversation.id, diff --git a/server/src/__tests__/fast-responses.test.ts b/server/src/__tests__/fast-responses.test.ts index 19eae5d692f..b08e4aa3a99 100644 --- a/server/src/__tests__/fast-responses.test.ts +++ b/server/src/__tests__/fast-responses.test.ts @@ -37,7 +37,7 @@ import { instanceSettingsService } from "../services/instance-settings.js"; import { settleConversationTurn } from "../services/agent-conversations.js"; import { buildLowTrustSourceTrust, LOW_TRUST_QUARANTINED_BODY } from "../services/source-trust.js"; import { deliverBudgetEnforcement } from "../services/budgets.js"; -let database: Awaited>, +let database: Awaited> | undefined, db: ReturnType, home: string; beforeAll(async () => { @@ -46,12 +46,15 @@ beforeAll(async () => { ); vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_INSTANCE_ID", "fast-response-fixture"); - database = await startEmbeddedPostgresTestDatabase( + const external = process.env.PAPERCLIP_TEST_DATABASE_URL; + if (external && !new URL(external).pathname.endsWith("_test")) throw new Error("Fast-response tests require a database ending in _test"); + database = external ? undefined : await startEmbeddedPostgresTestDatabase( "paperclip-fast-response-db-", ); - db = createDb(database.connectionString); + db = createDb(external ?? database!.connectionString); }, 90000); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await database?.cleanup(); vi.unstubAllEnvs(); if (home) await rm(home, { recursive: true, force: true }); diff --git a/server/src/app.ts b/server/src/app.ts index 865424a5c3e..daab82394e3 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1289,6 +1289,7 @@ export async function createApp( chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 2ac5ed4eed0..e9855e5cc5e 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -11,6 +11,7 @@ import { credentialFingerprint } from "./chat-channels/credential-fingerprint.js import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js"; +import { githubResponseCommentService } from "./chat-github-response-comments.js"; import { createGitHubMaintenance } from "./chat-channels/github/maintenance.js"; import { createSpekoChannelService, authorizeSpekoPublicVoiceTask, replaceSpekoChannelRuntime, SPEKO_CONVERSATION_IS_LINEAR } from "./chat-channels/speko/runtime.js"; import { prepareDiscordRuntimeCommand } from "./chat-channels/discord/runtime-commands.js"; @@ -3503,6 +3504,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { registerInteractionContinuationWork, + registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork, registerTerminalConfirmationWork, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices, diff --git a/server/src/services/chat-channels/github/acknowledgement-work.ts b/server/src/services/chat-channels/github/acknowledgement-work.ts new file mode 100644 index 00000000000..6bd60b0ed78 --- /dev/null +++ b/server/src/services/chat-channels/github/acknowledgement-work.ts @@ -0,0 +1,112 @@ +import { readGitHubResponseRun } from "./response-run-state.js"; +import { githubRunReplyState } from "../../chat-run-publications.js"; +import { and, asc, eq, gt, isNotNull, or, sql } from "drizzle-orm"; +import { chatActions, chatConversations, chatEndpointLeases, chatEndpointResources, + chatPublications, fastResponseRequests, chatEndpoints, chatMessageLinks, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import type { ChatResponseScope } from "../../chat-response-availability.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; + +type Scope = ChatResponseScope & { endpointId?: string; repositoryId?: string; number?: number }; +type Entry = { id: string; companyId: string; endpointId: string; deliveryId: string; + issueId: string | null; conversationId: string | null; repositoryId: string | null; number: number | null; dueAt: number }; +type FastResponse = { status: string; expiresAt: string; publicationState: string | null }; + +export function registerGitHubAcknowledgementWork( + db: Db, + coordinator: ReturnType, + acknowledge: (deliveryId: string, canContinue: () => boolean) => Promise, + canRun: () => boolean, +) { + const repositoryId = sql`${chatEndpointResources.metadata}->>'providerRepositoryId'`; + const number = sql`substring(${chatConversations.externalThreadId} + from '^github:[^:]+:(?:issue:)?([1-9][0-9]*)(?::rc:[1-9][0-9]*)?$')`; + const selection = { + id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId, + deliveryId: chatActions.deliveryId, conversationId: chatActions.conversationId, issueId: chatConversations.issueId, + agentId: chatEndpoints.assignedAgentId, + sourceMessageId: sql`(select m.comment_id from ${chatMessageLinks} m + where m.company_id = ${chatActions.companyId} and m.endpoint_id = ${chatActions.endpointId} + and m.delivery_id = ${chatActions.deliveryId} and m.direction = 'inbound' limit 1)`, + repositoryId, number, status: chatActions.status, result: chatActions.result, + leaseExpiresAt: sql`(select max(l.expires_at) from ${chatEndpointLeases} l + where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId} + and l.lease_key = 'github-publication:' || (${repositoryId}) || ':' || ${number})`, + fast: sql`(select jsonb_build_object('status', f.status, 'expiresAt', f.expires_at, + 'publicationState', p.state) from ${fastResponseRequests} f left join ${chatPublications} p + on p.company_id = f.company_id and p.comment_id = f.comment_id + and p.endpoint_id = f.endpoint_id and p.conversation_id = f.conversation_id + where f.company_id = ${chatActions.companyId} and f.endpoint_id = ${chatActions.endpointId} + and f.delivery_id = ${chatActions.deliveryId} limit 1)`, + }; + async function page(scope?: Scope, cursor?: string, actionId?: string) { + return db.select(selection).from(chatActions) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId))) + .leftJoin(chatConversations, and(eq(chatConversations.id, chatActions.conversationId), eq(chatConversations.companyId, chatActions.companyId))) + .leftJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), eq(chatEndpointResources.companyId, chatActions.companyId))) + .where(and(eq(chatActions.kind, "github_response_comment"), isNotNull(chatActions.deliveryId), + sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`, + or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)), + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(chatConversations.issueId, scope.issueId) : undefined, + scope?.conversationId ? eq(chatActions.conversationId, scope.conversationId) : undefined, + scope?.deliveryId ? eq(chatActions.deliveryId, scope.deliveryId) : undefined, + scope?.endpointId ? eq(chatActions.endpointId, scope.endpointId) : undefined, + scope?.repositoryId ? eq(repositoryId, scope.repositoryId) : undefined, + scope?.number ? eq(number, String(scope.number)) : undefined, + cursor ? gt(chatActions.id, cursor) : undefined, + actionId ? eq(chatActions.id, actionId) : undefined, + )).orderBy(asc(chatActions.id)).limit(100); + } + async function entry(row: Awaited>[number]): Promise { + const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, { + companyId: row.companyId, endpointId: row.endpointId, agentId: row.agentId, + issueId: row.issueId, deliveryId: row.deliveryId!, sourceMessageId: row.sourceMessageId, + }); + // Execution success alone does not decide whether a final response exists. + // Presentation/publication commits invalidate this parked request. + if (row.issueId && ownsLatestRequest && latestRun?.nativePresentationPending && + await githubRunReplyState(db, { companyId: row.companyId, endpointId: row.endpointId, + issueId: row.issueId, runId: latestRun.id }) === "none") return null; + let dueAt = row.status === "failed" ? Date.parse(String(row.result?.retryAt)) : 0; + if (!Number.isFinite(dueAt)) throw new Error("GitHub acknowledgement has an invalid retry deadline"); + const fast = row.fast; + if (fast?.publicationState === "streaming") return null; + if (fast && !["published", "delivery_unknown"].includes(fast.publicationState ?? "") && + (["pending", "running"].includes(fast.status) || ["pending", "retry"].includes(fast.publicationState ?? ""))) { + dueAt = Math.max(dueAt, Date.parse(fast.expiresAt)); + } + if (row.leaseExpiresAt) dueAt = Math.max(dueAt, Date.parse(row.leaseExpiresAt)); + return { id: row.id, companyId: row.companyId, endpointId: row.endpointId, deliveryId: row.deliveryId!, + issueId: row.issueId, conversationId: row.conversationId, repositoryId: row.repositoryId, number: row.number ? Number(row.number) : null, dueAt }; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubAcknowledgements, + decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (value: Entry, scope?: Scope) => !scope || (value.companyId === scope.companyId && + (!scope.issueId || value.issueId === scope.issueId) && (!scope.deliveryId || value.deliveryId === scope.deliveryId) && + (!scope.conversationId || value.conversationId === scope.conversationId) && + (!scope.endpointId || value.endpointId === scope.endpointId) && (!scope.repositoryId || value.repositoryId === scope.repositoryId) && + (!scope.number || value.number === scope.number)), + load: async (scope?: Scope, signal?: AbortSignal) => { + const entries: Entry[] = []; + let cursor: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await page(scope, cursor); + for (const row of rows) { const value = await entry(row); if (value) entries.push(value); } + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw signal?.reason ?? new Error("GitHub acknowledgement recovery stopped"); + }, + check: async (value: Entry, signal?: AbortSignal) => { + const [row] = await page(undefined, undefined, value.id); + const current = row ? await entry(row) : null; + if (!current || current.dueAt > Date.now() || !canRun() || signal?.aborted) return current; + await acknowledge(current.deliveryId, () => canRun() && !signal?.aborted); + const [remaining] = await page(undefined, undefined, value.id); + return remaining ? await entry(remaining) : null; + }, + }, canRun); +} diff --git a/server/src/services/chat-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts index 8d09defd0f3..5ae081cda53 100644 --- a/server/src/services/chat-channels/github/maintenance.ts +++ b/server/src/services/chat-channels/github/maintenance.ts @@ -13,7 +13,7 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet options.scheduleMessageProcessing(async () => { try { // Provider I/O must not block Slack, Teams, or ordinary message receipts. - await githubResponseCommentService(db, fetchImpl).processPending(); + await githubResponseCommentService(db, fetchImpl).processTerminalResponses(); await githubReviewCheckService(db, fetchImpl).processPending(); } finally { githubMaintenancePending = false; diff --git a/server/src/services/chat-channels/github/response-run-state.ts b/server/src/services/chat-channels/github/response-run-state.ts new file mode 100644 index 00000000000..2568c73d366 --- /dev/null +++ b/server/src/services/chat-channels/github/response-run-state.ts @@ -0,0 +1,71 @@ +import { and, desc, eq, or, sql } from "drizzle-orm"; +import { chatMessageLinks, heartbeatRuns, type Db } from "@paperclipai/db"; +import { chatRunSourceMessageId } from "../run-origin.js"; + +export const responseRunColumns = { + id: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + responsibleUserId: heartbeatRuns.responsibleUserId, + activeIdentityContextId: heartbeatRuns.activeIdentityContextId, + nativeIssueId: heartbeatRuns.nativeIssueId, + issueId: heartbeatRuns.issueId, + status: heartbeatRuns.status, + updatedAt: heartbeatRuns.updatedAt, + // Terminal execution can precede native presentation commit. Until its + // durable decision exists, missing output is undecided rather than failure. + nativePresentationPending: sql`coalesce( + ${heartbeatRuns.status} = 'succeeded' + and ${heartbeatRuns.runtimeMode} = 'native' + and (${heartbeatRuns.nativePhase} is distinct from 'committed' + or ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' is distinct from 'paperclip.run_presentation_decision.v1' + or coalesce(${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction', '') not in ('none', 'create', 'reuse')), + false + )`, + // Project only the server-committed no-publication decision, never the + // potentially large native result or model-authored summary. + passiveResponseWait: sql`coalesce( + ${heartbeatRuns.status} = 'succeeded' + and ${heartbeatRuns.runtimeMode} = 'native' + and ${heartbeatRuns.nativePhase} = 'committed' + and ${heartbeatRuns.resultJson}->>'finalizationPhase' = 'committed' + and ${heartbeatRuns.resultJson}->>'finalizationReasonCode' = 'external_chat_response_waiting' + and ${heartbeatRuns.resultJson}->'nativeResult'->>'schema' = 'paperclip.run_result.v1' + and ${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition' = 'yielded' + and ${heartbeatRuns.resultJson}->'nativeResult'->'continuation'->>'kind' = 'response_wake' + and ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' = 'paperclip.run_presentation_decision.v1' + and ${heartbeatRuns.resultJson}->'presentationDecision'->>'chosenSource' = 'none' + and ${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction' = 'none' + and ${heartbeatRuns.resultJson}->'presentationDecision'->'reasonCodes' @> '["yielded_control_plane_wait"]'::jsonb, + false + )`, + contextSnapshot: sql>`jsonb_build_object( + 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', + 'taskId', ${heartbeatRuns.contextSnapshot}->'taskId', + 'wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId', + 'wakeCommentIds', ${heartbeatRuns.contextSnapshot}->'wakeCommentIds' + )`, +}; +/** Shared request ownership and presentation state for delivery and recovery. */ +export async function readGitHubResponseRun(db: Db, input: { + companyId: string; endpointId: string; agentId: string; issueId: string | null; + deliveryId: string; sourceMessageId: string | null; +}) { + const [requestRun] = await db.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).innerJoin(chatMessageLinks, and( + eq(chatMessageLinks.companyId, input.companyId), eq(chatMessageLinks.endpointId, input.endpointId), + eq(chatMessageLinks.deliveryId, input.deliveryId), eq(chatMessageLinks.direction, "inbound"), + or(sql`${chatMessageLinks.commentId}::text = ${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`, + sql`coalesce(${heartbeatRuns.contextSnapshot}->'wakeCommentIds', '[]'::jsonb) ? ${chatMessageLinks.commentId}::text`), + )).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${input.issueId}`)) + .orderBy(desc(heartbeatRuns.createdAt)).limit(1); + const [latestRun] = await db.select(responseRunColumns).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId), + sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${input.issueId}`, + )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); + const originMessageId = latestRun?.activeIdentityContextId ? await chatRunSourceMessageId(db, latestRun, + { allowLegacyWakeComment: true, provider: "github" }) : null; + const ownsLatestRequest = !!latestRun && (requestRun?.runId === latestRun.id || + (!!input.sourceMessageId && originMessageId === input.sourceMessageId)); + return { latestRun, ownsLatestRequest }; +} diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index 8a4c8f7ed7a..5f3507d5473 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; @@ -121,6 +122,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur ? "cancelled" : "failed"; await notifyChatPublicationWork(tx, state === "failed" || state === "cancelled" ? publication : undefined); + if (state === "retry" || state === "delivery_unknown") await notifyChatResponseAvailability(tx, publication); const [ownedPublication] = await tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-channels/publications/queue.ts b/server/src/services/chat-channels/publications/queue.ts index a1b3b3e7bf3..f71f8725973 100644 --- a/server/src/services/chat-channels/publications/queue.ts +++ b/server/src/services/chat-channels/publications/queue.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; import { unresolvedSlackPublicationCondition } from "../../connectors/slack-publication.js"; @@ -242,9 +243,11 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti ), ), ); - const candidates = await tx.select({ id: chatPublications.id }).from(chatPublications).where(condition).for("update", { skipLocked: true }); + const candidates = await tx.select({ id: chatPublications.id, companyId: chatPublications.companyId, issueId: chatPublications.issueId }) + .from(chatPublications).where(condition).for("update", { skipLocked: true }); if (!candidates.length) return []; await notifyChatPublicationWork(tx); + for (const candidate of candidates) await notifyChatResponseAvailability(tx, candidate); return tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-channels/publications/task-control.ts b/server/src/services/chat-channels/publications/task-control.ts index ebe1a2b4ee9..040ea993dd6 100644 --- a/server/src/services/chat-channels/publications/task-control.ts +++ b/server/src/services/chat-channels/publications/task-control.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import type { DbOrTransaction } from "../runtime-types.js"; import { notifyChatPublicationWork } from "../../chat-work-notifications.js"; @@ -116,6 +117,7 @@ export function createChatPublicationTaskControl(db: Db) { // finishes after /new or /close. A racing insert that commits after this // transaction is rejected by the completed-conversation send guard. await notifyChatPublicationWork(tx); + await notifyChatResponseAvailability(tx, publication); await tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-channels/publications/worker.ts b/server/src/services/chat-channels/publications/worker.ts index 9f4f67c39ca..831a041fd39 100644 --- a/server/src/services/chat-channels/publications/worker.ts +++ b/server/src/services/chat-channels/publications/worker.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../../chat-response-availability.js"; import { notifyVoicePublication } from "../../voice/voice-reply-notifications.js"; import { createGitHubPublicationPolicy } from "../github/publication-policy.js"; @@ -155,6 +156,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp : Math.min(60_000, 2 ** Math.max(0, attempts) * 1_000); const result = await db.transaction(async (tx) => { await notifyChatPublicationWork(tx, terminal ? publication : undefined); + if (!terminal) await notifyChatResponseAvailability(tx, publication); const [settled] = await tx .update(chatPublications) .set({ @@ -475,6 +477,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp await publicationLease.assertOwned(); await db.transaction(async (tx) => { await notifyChatPublicationWork(tx); + await notifyChatResponseAvailability(tx, publication); return tx .update(chatPublications) .set({ diff --git a/server/src/services/chat-github-publication-lease.ts b/server/src/services/chat-github-publication-lease.ts index 36cf8a9880d..46d9e3b862b 100644 --- a/server/src/services/chat-github-publication-lease.ts +++ b/server/src/services/chat-github-publication-lease.ts @@ -116,6 +116,7 @@ export async function withGitHubPublicationLease( await renewal; await db.transaction(async tx => { await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify(scope)); await tx.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.id, claimed.id), eq(chatEndpointLeases.token, token))); }); } diff --git a/server/src/services/chat-github-response-comments.ts b/server/src/services/chat-github-response-comments.ts index 1c2b7383df2..fc1e1354f94 100644 --- a/server/src/services/chat-github-response-comments.ts +++ b/server/src/services/chat-github-response-comments.ts @@ -1,3 +1,4 @@ +import { readGitHubResponseRun, responseRunColumns } from "./chat-channels/github/response-run-state.js"; import { and, asc, desc, eq, inArray, or, sql } from "drizzle-orm"; import { chatActions, chatConversations, chatDeliveries, chatEndpoints, @@ -5,6 +6,9 @@ import { fastResponseRequests, chatPublications, type Db, } from "@paperclipai/db"; +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; +import { registerGitHubAcknowledgementWork } from "./chat-channels/github/acknowledgement-work.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { conflict, forbidden, HttpError } from "../errors.js"; import { githubChatPrincipalAccess } from "./chat-github-access.js"; import { githubAutomaticAdmission } from "./chat-github-events.js"; @@ -25,51 +29,8 @@ type Api = { request(path: string, options?: { method?: string; body?: unknow type Lease = { commit(write: (tx: Transaction) => Promise): Promise }; type Source = GitHubResponseSource; type Comment = { id: number; body?: string; html_url: string; user?: { login?: string } }; -const responseRunColumns = { - id: heartbeatRuns.id, - companyId: heartbeatRuns.companyId, - agentId: heartbeatRuns.agentId, - responsibleUserId: heartbeatRuns.responsibleUserId, - activeIdentityContextId: heartbeatRuns.activeIdentityContextId, - nativeIssueId: heartbeatRuns.nativeIssueId, - issueId: heartbeatRuns.issueId, - status: heartbeatRuns.status, - updatedAt: heartbeatRuns.updatedAt, - // Terminal execution can precede native presentation commit. Until its - // durable decision exists, missing output is undecided rather than failure. - nativePresentationPending: sql`coalesce( - ${heartbeatRuns.status} = 'succeeded' - and ${heartbeatRuns.runtimeMode} = 'native' - and (${heartbeatRuns.nativePhase} is distinct from 'committed' - or ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' is distinct from 'paperclip.run_presentation_decision.v1' - or coalesce(${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction', '') not in ('none', 'create', 'reuse')), - false - )`, - // Project only the server-committed no-publication decision, never the - // potentially large native result or model-authored summary. - passiveResponseWait: sql`coalesce( - ${heartbeatRuns.status} = 'succeeded' - and ${heartbeatRuns.runtimeMode} = 'native' - and ${heartbeatRuns.nativePhase} = 'committed' - and ${heartbeatRuns.resultJson}->>'finalizationPhase' = 'committed' - and ${heartbeatRuns.resultJson}->>'finalizationReasonCode' = 'external_chat_response_waiting' - and ${heartbeatRuns.resultJson}->'nativeResult'->>'schema' = 'paperclip.run_result.v1' - and ${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition' = 'yielded' - and ${heartbeatRuns.resultJson}->'nativeResult'->'continuation'->>'kind' = 'response_wake' - and ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' = 'paperclip.run_presentation_decision.v1' - and ${heartbeatRuns.resultJson}->'presentationDecision'->>'chosenSource' = 'none' - and ${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction' = 'none' - and ${heartbeatRuns.resultJson}->'presentationDecision'->'reasonCodes' @> '["yielded_control_plane_wait"]'::jsonb, - false - )`, - contextSnapshot: sql>`jsonb_build_object( - 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', - 'taskId', ${heartbeatRuns.contextSnapshot}->'taskId', - 'wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId', - 'wakeCommentIds', ${heartbeatRuns.contextSnapshot}->'wakeCommentIds' - )`, -}; const generation = (endpoint: Source["endpoint"]) => (endpoint.setup as { runtimeGeneration?: number }).runtimeGeneration ?? 0; +const stoppedAcknowledgement = Symbol("stopped GitHub acknowledgement"); const actionKey = (deliveryId: string) => `github_response:${deliveryId}`; const responseMarker = (source: Source) => isMainPullResponse(source) ? pullResponseMarker(source) : requestResponseMarker(source); @@ -91,18 +52,23 @@ async function findResponseComment(source: Source, api: Api, legacyMarkers: stri * Reconnect epochs fence publication attempts, not the durable comment identity. * Legacy receipts are also checked against the App-owned provider comment. */ async function stageResponse(db: Db, source: Source) { - const [inserted] = await db.insert(chatActions).values({ - companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - conversationId: source.conversation.id, deliveryId: source.delivery.id, - principalId: source.delivery.principalId, kind: "github_response_comment", - providerActionId: actionKey(source.delivery.id), - payload: { botExternalId: source.endpoint.botExternalId, agentId: source.endpoint.assignedAgentId }, - status: "received", - }).onConflictDoNothing().returning(); - const response = inserted ?? (await db.select().from(chatActions).where(and( + const lookup = () => db.select().from(chatActions).where(and( eq(chatActions.companyId, source.endpoint.companyId), eq(chatActions.endpointId, source.endpoint.id), eq(chatActions.providerActionId, actionKey(source.delivery.id)), - )))[0]; + )).then(rows => rows[0]); + const existing = await lookup(); + const response = existing ?? await db.transaction(async tx => { + await notifyChatResponseAvailability(tx, { companyId: source.endpoint.companyId, deliveryId: source.delivery.id }); + const [inserted] = await tx.insert(chatActions).values({ + companyId: source.endpoint.companyId, endpointId: source.endpoint.id, + conversationId: source.conversation.id, deliveryId: source.delivery.id, + principalId: source.delivery.principalId, kind: "github_response_comment", + providerActionId: actionKey(source.delivery.id), + payload: { botExternalId: source.endpoint.botExternalId, agentId: source.endpoint.assignedAgentId }, + status: "received", + }).onConflictDoNothing().returning(); + return inserted; + }) ?? await lookup(); if (!response || response.conversationId !== source.conversation.id || (response.payload.botExternalId !== undefined && response.payload.botExternalId !== source.endpoint.botExternalId) || response.payload.agentId !== source.endpoint.assignedAgentId) @@ -135,6 +101,10 @@ export async function writeGitHubResponseComment( (input.versionAt && Date.parse(String(response.result.versionAt)) > input.versionAt.getTime()))) { if (!input.acknowledgement && response.result.final === true && !input.final) throw conflict("This GitHub response is already final"); + if (["received", "failed"].includes(response.status)) await lease.commit(async tx => { + await tx.update(chatActions).set({ status: "processed", updatedAt: new Date() }) + .where(and(eq(chatActions.id, response.id), eq(chatActions.status, response.status))); + }); return { id: String(response.result.id), url: String(response.result.url) }; } const marker = responseMarker(source); @@ -213,13 +183,18 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { throw forbidden("Invalid GitHub request repository"); return { ...row, repositoryId, repository: match[1]!, number: Number(match[2]), replyId: match[3] ?? null }; } - async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise) { + async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise, canContinue = () => true) { + if (!canContinue()) return; const source = await context(deliveryId); // Persist before competing for the publication lane. A busy lane leaves // durable acknowledgement work for maintenance instead of losing it. await stageResponse(db, source); + if (!canContinue()) return; return withGitHubPublicationLease(db, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - repositoryId: source.repositoryId, number: source.number }, fetchImpl, async lease => { + repositoryId: source.repositoryId, number: source.number }, (input, init) => { + if (!canContinue()) throw stoppedAcknowledgement; + return fetchImpl(input, init); + }, async lease => { const assertCurrent = async () => { await assertRun?.(); const current = await context(deliveryId); @@ -234,22 +209,11 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { let preserveFinal = false; if (finalBody === undefined) { const existing = await stageResponse(db, source); - const [requestRun] = await db.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).innerJoin(chatMessageLinks, and( - eq(chatMessageLinks.companyId, source.endpoint.companyId), eq(chatMessageLinks.endpointId, source.endpoint.id), - eq(chatMessageLinks.deliveryId, deliveryId), eq(chatMessageLinks.direction, "inbound"), - or(sql`${chatMessageLinks.commentId}::text = ${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`, - sql`coalesce(${heartbeatRuns.contextSnapshot}->'wakeCommentIds', '[]'::jsonb) ? ${chatMessageLinks.commentId}::text`), - )).where(and(eq(heartbeatRuns.companyId, source.endpoint.companyId), eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), - sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${source.conversation.issueId}`)) - .orderBy(desc(heartbeatRuns.createdAt)).limit(1); - const [latestRun] = await db.select(responseRunColumns).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, source.endpoint.companyId), eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), - sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${source.conversation.issueId}`, - )).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1); - const originMessageId = latestRun?.activeIdentityContextId ? await chatRunSourceMessageId(db, latestRun, - { allowLegacyWakeComment: true, provider: "github" }) : null; - const ownsLatestRequest = !!latestRun && (requestRun?.runId === latestRun.id || - (!!source.sourceMessageId && originMessageId === source.sourceMessageId)); + const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, { + companyId: source.endpoint.companyId, endpointId: source.endpoint.id, + agentId: source.endpoint.assignedAgentId, issueId: source.conversation.issueId, + deliveryId, sourceMessageId: source.sourceMessageId, + }); requestRunning = ownsLatestRequest && latestRun?.status === "running"; const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status); const run = terminal && (ownsLatestRequest || source.issueStatus === "blocked") ? latestRun : null; @@ -289,7 +253,10 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { (await stagePullResponse(db, source)).result?.id === existing.result.id)) { // Rotate no-op maintenance checks so a page of newer queued asks // cannot hide other conversations that need terminal reconciliation. - await lease.commit(async tx => { await tx.update(chatActions).set({ updatedAt: new Date() }).where(eq(chatActions.id, existing.id)); }); + await lease.commit(async tx => { await tx.update(chatActions).set({ + status: ["received", "failed"].includes(existing.status) ? "processed" : existing.status, + updatedAt: new Date(), + }).where(and(eq(chatActions.id, existing.id), eq(chatActions.status, existing.status))); }); return { id: String(existing.result.id), url: String(existing.result.url) }; } if (finalBody === undefined) { @@ -439,6 +406,9 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`, sql`(${chatActions.result}->>'retryAt')::timestamptz <= now()`)))).orderBy(asc(chatActions.createdAt)).limit(limit); for (const row of rows) if (row.deliveryId) await tryAcknowledge(row.deliveryId); + return rows.length + await processTerminalResponses(limit); + } + async function processTerminalResponses(limit = 10) { // Recovery can stop execution before a milestone publication is staged. // Revisit acknowledged requests with terminal execution, even after a // restart; successful acknowledgement is not successful task completion. @@ -463,21 +433,28 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { or coalesce(r.context_snapshot->'wakeCommentIds', '[]'::jsonb) ? l.comment_id::text))))`, )).orderBy(asc(chatActions.updatedAt)).limit(limit); for (const row of stopped) if (row.deliveryId) await tryAcknowledge(row.deliveryId); - return rows.length + stopped.length; + return stopped.length; } - async function tryAcknowledge(deliveryId: string) { - try { return await acknowledge(deliveryId); } + async function tryAcknowledge(deliveryId: string, canContinue = () => true) { + try { return await acknowledge(deliveryId, undefined, undefined, canContinue); } catch (error) { + if (error === stoppedAcknowledgement || !canContinue()) return; const [row] = await db.select().from(chatActions).where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.deliveryId, deliveryId))); if (!row || row.result?.final === true) return; const attempts = Number(row.result?.attempts ?? 0) + 1; const denied = error instanceof Error && "status" in error && error.status === 403; - await db.update(chatActions).set({ status: denied ? "cancelled" : "failed", result: { - ...row.result, attempts, retryable: !denied && attempts < 8, - retryAt: new Date(Date.now() + Math.min(300000, 1000 * 2 ** attempts)).toISOString(), - code: denied ? "authorization_changed" : "acknowledgement_unconfirmed", - }, updatedAt: new Date() }).where(and(eq(chatActions.id, row.id), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); + await db.transaction(async tx => { + await notifyChatResponseAvailability(tx, { companyId: row.companyId, deliveryId }); + await tx.update(chatActions).set({ status: denied ? "cancelled" : "failed", result: { + ...row.result, attempts, retryable: !denied && attempts < 8, + retryAt: new Date(Date.now() + Math.min(300000, 1000 * 2 ** attempts)).toISOString(), + code: denied ? "authorization_changed" : "acknowledgement_unconfirmed", + }, updatedAt: new Date() }).where(and(eq(chatActions.id, row.id), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); + }); } } - return { tryAcknowledge, processPending, failRun, finishRun, hasWorkingResponseForRun }; + return { tryAcknowledge, processPending, processTerminalResponses, failRun, finishRun, hasWorkingResponseForRun, + registerAcknowledgementWork: (coordinator: ReturnType, canRun: () => boolean) => + registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun), + }; } diff --git a/server/src/services/chat-response-availability.ts b/server/src/services/chat-response-availability.ts new file mode 100644 index 00000000000..29de7ca303e --- /dev/null +++ b/server/src/services/chat-response-availability.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string }; + +export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); +} diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 972d6789bb7..1727d2eeca3 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatConfirmationWork } from "./chat-confirmation-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; @@ -15,4 +16,6 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston // Admission and current chat authority also govern saved queue notices. await notifyChatInboundWakeNotices(tx, scope); await notifyChatConfirmationWork(tx, scope); + // Native presentation and run ownership can unblock a parked acknowledgement. + await notifyChatResponseAvailability(tx, scope); } diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 5e0d05262c9..9b878913e8d 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -17,6 +18,7 @@ export async function notifyChatPublicationWork( // release its provider response slot. Callers supply the affected issue only // for those changes; transport claim/lease bookkeeping has no projection debt. if (milestoneScope?.issueId) { + await notifyChatResponseAvailability(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId }); await notifyChatRunMilestones(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId }); } } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 428bc88b2ca..cab6f3b8c94 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -41,6 +41,7 @@ export const DELIVERY_QUEUES = { chatProviderEffects: "chat-provider-effects", chatGitHubIngress: "chat-github-ingress", chatGitHubPublications: "chat-github-publications", + chatGitHubAcknowledgements: "chat-github-acknowledgements", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", chatReceivers: "chat-receivers", diff --git a/server/src/services/fast-responses.ts b/server/src/services/fast-responses.ts index 8ba6665dea6..4f0d2dddb6d 100644 --- a/server/src/services/fast-responses.ts +++ b/server/src/services/fast-responses.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { insertIssueComment } from "./issue-comment-writes.js"; import { randomUUID } from "node:crypto"; import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js"; @@ -195,6 +196,7 @@ export async function enqueueFastResponse(tx: Db, source: FastResponseSource) { if (source.acceptedAt.getTime() + FAST_RESPONSE_DEADLINE_MS <= Date.now()) return; await notifyDeliveryWork(tx, DELIVERY_QUEUES.fastResponse); + if (source.endpointId && source.deliveryId) await notifyChatResponseAvailability(tx, { companyId: source.companyId, deliveryId: source.deliveryId }); await tx .insert(fastResponseRequests) .values({ @@ -549,20 +551,23 @@ export function fastResponseService( } } async function skip(request: Request, reason: string) { - await db - .update(fastResponseRequests) - .set({ - status: "skipped", - publicationStatus: "suppressed", - errorCode: reason, - finishedAt: new Date(), - }) - .where( - and( - eq(fastResponseRequests.id, request.id), - eq(fastResponseRequests.status, "pending"), - ), - ); + await db.transaction(async tx => { + if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId }); + await tx + .update(fastResponseRequests) + .set({ + status: "skipped", + publicationStatus: "suppressed", + errorCode: reason, + finishedAt: new Date(), + }) + .where( + and( + eq(fastResponseRequests.id, request.id), + eq(fastResponseRequests.status, "pending"), + ), + ); + }); } async function settle(request: Request, outcome: FastResponseOutcome) { await withBudgetSpan("budget.accounting.settlement", () => withAccountingTransaction( @@ -625,6 +630,7 @@ export function fastResponseService( }, ); const finishedAt = new Date(); + if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId }); await tx .update(fastResponseRequests) .set({ @@ -993,6 +999,7 @@ export function fastResponseService( idempotencyKey: `fast-response:${request.id}`, payload: safe, }); + if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId }); await tx .update(fastResponseRequests) .set({ diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 62a0396e2e1..1e5e13b858c 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -346,6 +346,19 @@ function controllerLeaseLossResultJson(signal: AbortSignal) { : {}; } +/** Persist the resolved presentation and wake dependents in its committing transaction. */ +export async function recordRunPresentationDecision(db: Db, input: { + companyId: string; runId: string; issueId?: string; decision: RunPresentationDecision; +}) { + await db.transaction(async tx => { + await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId }); + await tx.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: input.decision })}::jsonb`, + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, input.runId)); + }); +} + export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunCompletionDependencies) { const { getRun, @@ -892,13 +905,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC }; } - await db.transaction(async tx => { - await notifyChatRunMilestones(tx, { companyId: livenessRun.companyId, issueId: issueId ?? undefined }); - await tx.update(heartbeatRuns).set({ - resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision })}::jsonb`, - updatedAt: new Date(), - }).where(eq(heartbeatRuns.id, livenessRun.id)); - }); + await recordRunPresentationDecision(db, { companyId: livenessRun.companyId, + runId: livenessRun.id, issueId: issueId ?? undefined, decision: presentationDecision }); await appendRunEvent(livenessRun, { eventType: "run.presentation.resolved", stream: "system",