From f8226b0515634f9c7c74e80bb6cf18f3c2634c12 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 16:51:57 -0500 Subject: [PATCH] fix(server): wake GitHub terminal responses from committed changes (#17661) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit https://entire.com/gh/paperclipai/-/trails/210/changes/paperclip/210 ## Thinking Path > - Paperclip coordinates agent work and presents results in chat. > - GitHub acknowledgements already persist the response action and provider receipt. > - A terminal run can still owe a final response after acknowledgement. > - Periodic maintenance searches those rows even when no relevant state changed. > - Committed run, issue, presentation, lease and publication changes identify when recovery can advance. > - The owner should recover from those events and real deadlines, then become quiet. ## Linked Issues or Issue Description Refs #17656, which is merged. This existing terminal-response slice targets master. **Problem:** A saved GitHub response can remain nonfinal while its run, native presentation, publication or source is unresolved. Removing maintenance without covering those producers loses recovery. Polling settled history keeps the tenant active. **Expected behavior:** Committed supported changes wake the terminal owner. It reconstructs existing actions, preserves provider authority and pending publication ownership, waits at saved deadlines where needed, and becomes quiet after settlement. ## What Changed - Register terminal GitHub response work with the shared scoped deadline coordinator, using existing action/delivery/publication/run rows. - Notify response availability from supported run milestones, native identity, recovery resolution, issue updates and tree/workspace restoration writes. Preserve outer transaction commit/rollback. - Keep GitHub response interpretation in its platform owner. Use the bounded run projection for both direct authorization and scheduling, including native presentation guards and deliberate passive outcomes. - Retain the existing publication lease, markers, source/principal checks and final receipt precedence. A publication owner or uncertain outcome keeps its existing authority; the worker does not replay it. - Remove only recurring terminal-response recovery, preserving the manual path and neighbouring duties. Add no table or scheduler service. - Regenerate the original sparse terminal-work index as `0401_yummy_ultimates.sql`, concurrently and idempotently through the existing top-level migration runner. Preserve all prior SQL/journal entries and the five newest snapshots. ## Verification Published head: `88a209e28b969b5229e9a621aeddfd3cbae9b0cd`, integrated with landed #17656 (`ba5bba86dcdde32b14d4d7dec1b51898e5f6f22d`) and master `52b86c2032a189b0ea259c5b0cb7158b07ef788c`. - **74 distinct current cases pass:** 17 terminal-response lifecycle/producer cases (1,721 other chat cases skipped), all six queued-comment consumption regressions for the new upstream query optimization, all 30 heartbeat completion cases, all eight real/unit issue-tree control cases and all 13 startup cases. Database build, server typecheck/build, module boundaries and four fresh migrations through 0401 pass. - The earlier **304-case** prepared-parent matrix passes at `aa72f32641b2e2c680342cd49477272349aca9db`: all 253 GitHub workflow cases, 30 completion cases, eight issue-tree cases and 13 startup cases. It retains that revision scope. Repeated cases are not added to the current 74. - The terminal matrix covers finalized-history quiet, deleted Board sources, pending presentation/passive outcomes, outer rollback/commit, paused/new runs, restart/drain/standby, committed native identity, lease waits, full-page unrelated history, uncertain publication ownership and tree/workspace/exhausted-recovery producers. - A populated 0400 database upgrades to 0401 with all 399 prior migration receipts identical, exactly one added, and none added on replay. The terminal index is valid/ready on fresh and upgraded databases. Its SQL matches the original draft after whitespace normalization; generation keeps snapshots 0397–0401. - Normal integration `3c8339a541bf68bc9957b9d5a8df80d3a7c3452d` preserves the complete 24-path owned binary patch and all five new master paths. The landed parent tree also equals the previously published parent tree byte for byte. Both normal merge parents and the exact composition tree are checked. No force push or rebase occurs. - [The owner audit](https://github.com/paperclipai/paperclip/blob/88a209e28b969b5229e9a621aeddfd3cbae9b0cd/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md#acceptance-on-the-landed-github-parent-and-current-master) preserves the original producer, quietness, failure-isolation, skipped-suite and presentation-wait results with their scopes. Current passes do not erase earlier failures or turn skips into passes. - The audit-only follow-up 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 suite, live GitHub tenant or deployed sleep result is claimed. ## Risks - Process-local notifications cover supported application writers and shared pools. Independent SQL writers and other replicas gain no new wake guarantee. - Native terminal status alone does not authorize fallback publication. Pending presentation and existing publication ownership remain separate obligations. - Stop can leave an in-flight provider request settling; admission fences stop a subsequent request. Unknown outcomes retain their existing recovery rules. - Index creation can take time on large histories. Concurrent/idempotent migration and existing replay recovery remain in charge. - Worker quiet does not establish host/database suspension. Cloud alarm coverage and sleep admission remain unchanged. ## 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 | 29 ++ ...25-github-terminal-response-sleep-audit.md | 379 +++++++++++++++++ packages/db/src/client.ts | 1 + .../src/migrations/0401_yummy_ultimates.sql | 3 + ...{0396_snapshot.json => 0401_snapshot.json} | 164 +++++++- packages/db/src/migrations/meta/_journal.json | 7 + packages/db/src/schema/chat_channels.ts | 3 + .../app-private-hostname-gate.test.ts | 3 +- .../chat-channels.integration.test.ts | 388 ++++++++++++++++++ server/src/app.ts | 1 + server/src/services/chat-channels.ts | 1 + .../chat-channels/github/maintenance.ts | 2 - .../github/response-execution.ts | 42 ++ .../github/terminal-response-work.ts | 106 +++++ .../services/chat-github-publication-lease.ts | 1 + .../services/chat-github-response-comments.ts | 85 ++-- server/src/services/chat-github-reviews.ts | 3 + .../services/chat-response-availability.ts | 1 + .../chat-run-milestone-notifications.ts | 1 + .../services/delivery-work-notifications.ts | 1 + .../services/execution-recovery-resolution.ts | 2 + server/src/services/heartbeat/queue.ts | 2 + server/src/services/issue-tree-control.ts | 6 +- server/src/services/issues.ts | 5 + 24 files changed, 1185 insertions(+), 51 deletions(-) create mode 100644 doc/plans/2026-10-25-github-terminal-response-sleep-audit.md create mode 100644 packages/db/src/migrations/0401_yummy_ultimates.sql rename packages/db/src/migrations/meta/{0396_snapshot.json => 0401_snapshot.json} (99%) create mode 100644 server/src/services/chat-channels/github/response-execution.ts create mode 100644 server/src/services/chat-channels/github/terminal-response-work.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 5a2d80fa0a4..10cc1ee7681 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2369,3 +2369,32 @@ 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. + +## Slice 34 accepted head and slice 35 checkpoint + +[#17656](https://github.com/paperclipai/paperclip/pull/17656) at +`43900c1f94500af4c4de1987f9608ac4e12feae6` passes all 53 remote checks, +fresh Greptile 5/5, and has zero review threads. Final chat/publication/startup +acceptance passes all 1,557 tests with no skips in 514.70 seconds. Workspace +types/build and 69 fast-response cases pass. The PR remains draft and unmerged. + +Branch `codex/sleep-35-github-terminal-responses` starts from that head. +It extracts provider-owned terminal-state decisions, records successful paused +presentation versions in existing receipts, and adds a scoped worker plus +committed issue/run/response-slot notifications. Eight new real-database +lifecycle cases and nine existing owner cases pass; server types/build pass. +The broader GitHub workflow suite is running. Startup registration and polling +removal wait for the three identified direct issue writers and remaining audit +and acceptance requirements in the terminal-response audit. No slice-35 PR or +full-stack readiness is claimed yet. Earlier parent CI/review findings, all other +inventory duties, and reconciliation with current master remain required. + +Slice 35 now covers the reproduced direct tree-restoration, workspace-admission +and exhausted-recovery producers. Real publication-owner tests preserve both +known-unsent cancellation and ambiguous outcome parking. Startup registers the +new terminal worker and recurring GitHub maintenance no longer invokes that duty. +Fourteen new worker cases plus one existing case pass; workspace types/build and +migration application/replay/safety pass. The partial candidate index excludes +10,000 completed rows in a rollback-only query-plan experiment. Fresh full +chat/publication/startup acceptance is running; remote gates remain required. +See the terminal-response audit for all retained failures and scoped evidence. diff --git a/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md b/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md new file mode 100644 index 00000000000..e908edceb58 --- /dev/null +++ b/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md @@ -0,0 +1,379 @@ +# GitHub terminal-response recovery + +Slice 35 starts from acknowledgement PR #17656 at +`43900c1f94500af4c4de1987f9608ac4e12feae6`. +The final parent acceptance passed all 1,557 chat/publication/startup tests without +skips in 514.70 seconds. Parent workspace types/build and 69 fast-response tests +also passed. Greptile reviewed that head at 5/5 without findings. Remote CI was +still completing when this slice started; no merge or full-stack completion is +claimed. + +## Obligation and ownership + +A received acknowledgement and an owed terminal response are different states +of the existing `github_response_comment` action. Slice 34 owns received actions +and retryable failed deliveries. This slice owns processed, nonfinal actions +whose bound request has a terminal outcome or needs a paused notice. Review-check +publication and stale-head projection repair remain independent duties. + +The provider owner must determine exact source/run provenance, latest-run +ownership, and whether a real reply is confirmed or uncertain. Keep those queries +and presentation decisions in `chat-channels/github/response-execution.ts`; shared +notification and scheduler code must not interpret GitHub payloads. Existing +principal/configuration checks, provider markers and repository publication leases +remain the authority for sends and receipt writes. + +## Passive history is not unfinished delivery + +A paused response deliberately remains nonfinal so a later resumed run can answer +it. The old scan repeatedly revisits it, even when it makes no provider mutation. +The replacement needs a saved paused-presentation version in the existing receipt, +written only with confirmed receipt persistence. A changed run or pause state +invalidates that version. No table or separate queue is needed. + +An unrelated newer terminal run does not automatically authorize closing every +older request. Resolve native identity and legacy wake-comment provenance using +the existing owner. Passive non-owning rows must be excluded without acquiring +and releasing a publication lease on every wake. That would otherwise create a +self-wake loop through the release notification. + +## Producer audit still required + +- Run and issue transitions, identity/lineage changes and restored authority use + the existing milestone notification boundary; extend it once with the narrow + response-state topic after verifying coverage. +- Response action completion and provider publication/assessment settlement must + wake affected terminal candidates. Repository lease release is scoped to its + company, endpoint, repository and PR/issue number. +- An unresolved competing lease owns its saved expiration, not a permanent scan. +- Failed terminal sends remain with the existing acknowledgement retry owner, + including the current attempt budget and ambiguous-outcome policy. +- Startup reconstructs eligible work; drain/standby/shutdown retain skipped work + and prevent starting subsequent provider requests. + +## Acceptance required before polling removal + +Use real PostgreSQL and the existing controlled GitHub fixture. Cover terminal +run producer commit/rollback, native identity, blocked-but-non-owning requests, +paused-history quiet and resume, final/superseded history, lease release and +restart, uncertain reply settlement, admission changes between provider requests, +changes during a sweep, and candidate prefixes across the bounded page size. +Measure actual SQL and HTTP effects while idle, not only timer registration. + +Current state: extracting the provider-owned execution decision and persisting +confirmed paused presentation versions. The terminal worker, complete producer +audit, lifecycle tests, startup wiring and recurring-call removal remain pending. + +## Implemented worker checkpoint + +The provider execution resolver is extracted. Successful paused receipt writes +save `pausedVersion` in the existing result. The scoped worker reconstructs +processed nonfinal candidates in 100-row ID pages, excludes unrelated run +history and unresolved reply owners, and uses real publication-lease expiration +when another writer owns the lane. A settled paused receipt is passive. New +run state invalidates it. The owning acknowledgement service still checks +current authority and stops subsequent provider requests at admission/shutdown. +Failed sends transfer to the existing acknowledgement retry worker. + +The response-availability helper now wakes this worker as well as initial +acknowledgements. The milestone boundary wakes it for existing run, identity, +lineage and authority producers. Issue-service status/assignee changes notify +inside their owning transaction. GitHub publication staging, missing-binding +cancellation and scoped lease release also notify. Registration is exposed by +the provider service for tests, but app startup still does not register it and +the old terminal maintenance call remains until the audit below is complete. + +New acceptance covers eight cases: startup terminal recovery and settled-history +quiet; outer issue transaction rollback/commit; paused-history quiet and a later +run; restart; drain; standby; native-identity rollback/commit; real repository +lease contention/release. All eight pass in `/tmp/sleep35-lifecycle-coalesced.log` +(21.88 seconds). The SQL-quiet checks advance 120 seconds and inspect real +`select` and `transaction` calls. GitHub HTTP remains controlled. + +Server typecheck and build pass at this checkpoint, with logs +`/tmp/sleep35-final-server-types.log` and `/tmp/sleep35-final-server-build.log`. +All GitHub review-workflow acceptance is running in +`/tmp/sleep35-github-acceptance.log`; no broad result is claimed yet. + +Material failures retained: + +- `/tmp/sleep35-before-producer.log`: all three first cases failed because the + test coordinator omitted its required `canRun` function. Corrected the fixture. +- `/tmp/sleep35-before-producer-fixed-fixture.log`: startup and paused recovery + passed, while the real issue commit failed to wake terminal response work. + The issue owner now signals the response slot before writing its status. +- `/tmp/sleep35-producer-fixed.log`: the producer regression and 15 adjacent + cases passed; the paused restart assertion observed the legitimate pending + coalesced lease-release pass rather than waiting for settlement. +- `/tmp/sleep35-lifecycle.log`: four passed and four assertions failed at the + default one-second poll timeout or before the coalesced pass. The coordinator + itself coalesces at one second. Explicit five-second observation bounds now + allow that existing behavior; `/tmp/sleep35-lifecycle-coalesced.log` passes all + eight. No production delay or retry policy was widened to make tests pass. +- `/tmp/sleep35-owner-extraction.log`: nine existing ownership/paused/native/ + legacy-placeholder cases passed after extraction. + +## Direct issue writers found before startup wiring + +These are unresolved producer-audit items, not exemptions: + +1. `issue-tree-control.ts::restoreIssueStatusesForHold` restores issue status + directly. Its cancellation path already calls `issueService.update`, but the + restoration path does not. Test a restored blocked request with existing + response history and cover its transaction before removing the old sweep. +2. `heartbeat/queue.ts` directly blocks a task whose isolated worktree lacks a + usable project. This can happen before any run is admitted, so a run-terminal + notification cannot cover it. Existing assembled coverage is in + `heartbeat-workspace-branch-containment.test.ts` around the projectless + isolated-worktree rejection case. Verify this producer's response notification. +3. `execution-recovery-resolution.ts::settleUnrecoverableExecutions` can set + status blocked after a run was already terminal. It does not currently send + the response notification. Existing real-database coverage is in + `native-safe-replacement.test.ts` and run-dispatch adapter tests. Preserve its + task/coordinator/run lock order and failure rollback when adding the signal. + +The native-finalization reconciler's blocked transition calls issueService and +is covered by the new owning boundary. Recovery/service's inspected blocked +transitions also call issueService. Runner goal statuses are session-goal data, +not issue status, and do not independently change this response predicate. +Additional dynamic status writers, lookup/index cost, uncertain-publication +settlement, page fairness, final startup wiring, full acceptance and remote review +still need completion. Do not present this checkpoint as polling removal. + +## Production registration and direct-writer repairs + +The three direct writers above now notify response availability before their +writes in the existing transactions. Three real-producer regressions reproduce +missing paused response settlement before the change. Tree cancellation continues +to use issueService; restoration now notifies each restored issue. Queue admission +notifies only a transition into blocked. Exhausted recovery does likewise and +retains its existing lock order and failure behavior. + +`/tmp/sleep35-direct-producers-before.log` records the missing tree/recovery wakes. +Its workspace case initially failed earlier because the fixture had not enabled +isolated workspaces, so it exercised normal admission instead of the intended +pre-dispatch refusal. The corrected fixture enables and restores the feature, +drains any executions on cleanup, and reproduces the missing workspace wake in +`/tmp/sleep35-workspace-producer-before.log`. The real issue state reaches blocked +before the missing-notice assertion, so the red result establishes the producer +boundary rather than an unrelated setup failure. All eleven producer/lifecycle +cases pass after repair in `/tmp/sleep35-direct-producers-fixed.log`. + +The remaining inspected direct status writes in agent-conversations, Slack +lifecycle/state, access reassignment, checkout/release, and native workspace +reconstruction move away from blocked or remove assignment. They do not create +a new paused-response obligation. Execution eligibility is reread before any +send, so an outstanding lease-delayed candidate observes these changes. New +run/lineage changes still go through the milestone boundary. Generic issue-service +assignment/status mutations notify directly. Agent deletion does not restore an +owed response. Goal status fields are separate session state. + +Two further cases drive the real review-publication owner through token acquisition, +a terminal run transition and pre-send authority denial. A known-unsent publication +cancels and wakes terminal fallback through lease release. A saved ambiguous write +remains parked, with no repeated SQL over 120 seconds. The latter does not assert +that the provider write failed or relax host safety. Both pass in +`/tmp/sleep35-publication-settlement.log`. The existing eight-attempt acknowledgement +policy remains the owner of actual failed fallback sends. + +The app now registers terminal-response work with warm-standby and idle-drain +gates, and recurring GitHub maintenance no longer calls terminal reconciliation. +The explicit/manual combined recovery entry point is preserved. Review-check +publication and stale-head review repair remain separate duties in maintenance. + +## Candidate lookup and final acceptance + +Generated migration `0395_jittery_norrin_radd.sql` adds a partial index for +processed, nonfinal response actions. It runs concurrently through the existing +top-level migration registry and is idempotent on replay. Snapshot pruning retains +0387 through 0391, exactly five. Application, replay, numbering/safety checks and +PostgreSQL index validity pass. No table or queue storage was added. + +The worker pages candidates by ID, then asks the GitHub owner to resolve actual +run provenance and reply state. This has per-candidate read cost at startup or +an affected scope change; it does not scan while settled. Issue/delivery/repository +hints narrow normal wakes. A 101-row prefix of unrelated retained requests cannot +hide the owed terminal response and does not publish those unrelated requests. +The shared helper provides fair 25-entry dispatch batches after reconstruction. + +A rollback-only experiment inserted 10,000 completed actions. The candidate index +query read four eligible rows using an index-only scan and four shared buffers. +See `/tmp/sleep35-index-plan.sql` and `/tmp/sleep35-index-plan.log`. This proves +completed-history exclusion for the candidate access path, not whole-worker latency +or fleet savings. The lifecycle tests prove settled SQL quiet separately. + +Verification: + +- The earlier extracted-owner GitHub acceptance passed 194 cases in 140.71 seconds + (`/tmp/sleep35-github-acceptance.log`) before the three direct-writer repairs. +- Final focused selection passes 15 cases: fourteen new terminal-worker cases + plus one existing matching case (`/tmp/sleep35-wired-focused.log`, 46.15 seconds). + The startup source suite did not match this focused selector and is included + in the full run below; it is not claimed passed by that focused command. +- Workspace typecheck and build pass after final source/wiring/index changes + (`/tmp/sleep35-workspace-types.log`, `/tmp/sleep35-workspace-build.log`). +- Fresh complete chat/publication/startup acceptance is running on + `paperclip_sleep35_full_20261025_test`, log `/tmp/sleep35-full-chat.log`. + Its expected count is 1,571. Do not claim completion before its terminal result. +- Remote checks/review and any resulting repairs remain required. + +Independent SQL writers and multiple replicas gain no new wake protocol. Cloud +alarm coverage and sleep admission are unchanged. The full recurring-server stack, +older failed gates/review findings and final master reconciliation remain open. + +## Quietness fixture settlement and isolation (October 26) + +The coordinator removes a due timer before it runs reconciliation. Therefore, +`nextWakeAt() === null` alone does not establish that the worker has finished. +The completed-history test now waits for both no scheduled deadline and zero +tracked active work before starting its existing two-minute quiet measurement. +It still requires zero database reads, zero transactions, and no provider writes. +No production worker, deadline, or recovery policy changes in this repair. + +The same test previously restored its database spies only after all assertions +passed. A failed quietness assertion left the transaction spy installed. The +later audit fixture bound that existing spy, then replaced its implementation +with a callback through the bound spy, causing recursive calls. + +A controlled experiment inserted one real SQL read inside the quiet measurement. +Before cleanup was repaired, the quietness assertion failed and the subsequent +`audit_insert` case failed with a stack overflow instead of its expected audit +error (two failures and one pass). With the same intentional SQL read and +unconditional spy restoration, only the quietness assertion failed; both audit +failure cases passed. The temporary SQL read was removed before acceptance. +This proves failure isolation without weakening the quietness assertion. It does +not independently reproduce the original CI timing race. + +All 228 cases in the complete GitHub agent review, Discord automatic command +registration, and resource-change audit groups then passed against a migrated, +disposable PostgreSQL database. Other groups were excluded by the test selector; +this is not a complete chat-file result. Logs: +`/tmp/sleep-chat-quiet-contamination-red.log`, +`/tmp/sleep-chat-quiet-contamination-contained.log`, and +`/tmp/sleep-chat-quiet-owner-tests.log`. + + +## Master integration: native presentation and passive review waits + +Master added two durable presentation guards after this worker was extracted. +The integrated provider resolver retains both through the same bounded run +projection used by direct response authorization. A terminal native status with +an unsettled presentation decision creates no fallback-send obligation yet. +The worker becomes quiet and the existing native finalizer milestone signal +reactivates it after a committed presentation update. A confirmed publication +still takes precedence over an absent fallback. + +For a main PR response whose exact owned run deliberately yielded without a +publication, acknowledgement retires its own action under the publication lease +and keeps the shared assessment untouched. Issue-thread and review-thread +fallbacks retain their existing behavior. The cancellation signals response +availability inside the committing transaction. No provider call is replayed. + +Acceptance includes master's passive-assessment and malformed/pending-decision +cases plus two worker cases: quiet pending presentation, rolled-back settlement, +committed passive or missing-response settlement, and final drain. The fixture +uses real PostgreSQL with controlled GitHub responses. The new cases must pass +before this integration is published; source merging alone is not acceptance. + + +## Current local preparation — October 28 + +Normal integration `ed1c9fcfa10d492bb56ea2e23cd97cddeebe9d5b` adopts the prepared +GitHub acknowledgement parent `3fef8f08c0337301706bc6348a05bb2a2fc50679`. +Every original domain/schema edit and unrelated parent blob is preserved. +The only conflict is a historical snapshot rename/delete; it retains the current +parent's 0396 snapshot byte for byte. The parent migration/client state is +identical before this slice's index is regenerated. Proof: +`/tmp/sleep35-parent-composition-proof.json`. + +`706cb6dbf58bf42a5f611319994c2c3907c9a80f` regenerates the original terminal +work index as `0401_yummy_ultimates.sql`. Its concurrent/idempotent index SQL is +identical after whitespace normalization to the old unlanded 0395 definition. +All prior SQL/journal entries and retained snapshot bytes remain unchanged. +Canonical generation keeps the five newest snapshots, 0397–0401. The existing +top-level migration registry uses the new filename. + +The exact parent-to-head migration-order gate, database build and production +server typechecking pass on that revision. Commands/receipts: +`/tmp/sleep35-current-compile-command.json` and +`/tmp/sleep35-current-{order,db-build,types}.*`. +These are compilation/history checks, not current lifecycle or PostgreSQL +migration-application acceptance. No current test count is claimed here. +The earlier lifecycle, quietness, provider, failure-isolation and presentation +results above retain their original revision scopes. + +This remains local and unpublished while preceding PRs finish their required +gates and merges. Actual landed-parent reconciliation, relevant functional +acceptance, fresh exact-head CI and Greptile 5/5 without actionable findings are +still required. No full local repository, live provider or deployed sleep result +is inferred from this preparation. The following audit-only commit leaves the +compiled source and migration bytes unchanged. + + +## Current prepared-parent functional acceptance + +Normal integration `aa72f32641b2e2c680342cd49477272349aca9db` adopts published +GitHub acknowledgement parent `09ca00b9f029838434ec042cda691345d67b493e`. +The complete original owned binary patch and all inherited master changes are +preserved. Proof: `/tmp/sleep35-published-parent-composition-proof.json`. + +**304 distinct cases pass:** all 253 GitHub agent-review workflow cases (1,485 +other chat cases skipped), all 30 heartbeat completion cases, all eight cases +across the complete real/unit issue-tree control suites, and all 13 startup +cases. Database build, server typechecking/build, module boundaries and three +fresh PostgreSQL migrations through 0401 pass. The control suite is explicitly +premigrated, preserving its original setup budget. Commands and actual fixtures: +`/tmp/sleep35-current-acceptance-command.json`; receipts/logs: +`/tmp/sleep35-current-{db-build,types,build,boundaries,github,completion,tree,startup}.*` +and `/tmp/sleep35-current-migrate-{github,completion,tree}.*`. +These results retain that prepared-parent revision scope. + + +## Acceptance on the landed GitHub parent and current master + +GitHub acknowledgements #17656 landed at +`ba5bba86dcdde32b14d4d7dec1b51898e5f6f22d`. Its entire tree is byte-identical +to the published parent `09ca00b9f029838434ec042cda691345d67b493e`. +Master then added the queued-comment consumption optimization #18021. +Normal two-parent integration `3c8339a541bf68bc9957b9d5a8df80d3a7c3452d` +adopts master `52b86c2032a189b0ea259c5b0cb7158b07ef788c`. It preserves the full +24-path owned binary patch and all five new master paths. Both normal parents +and the exact tree are checked, including historical squash-ancestry resolution +against the verified published parent. Proof: +`/tmp/sleep35-landed-parent-publication-proof.json`. No force push or rebase occurs. + +**74 distinct current cases pass:** all 17 terminal-response lifecycle/producer +cases (1,721 other chat cases skipped), all six queued-comment consumption +regressions for the new upstream optimization, all 30 heartbeat completion cases, +all eight real/unit issue-tree control cases and all 13 startup cases. Database +build, server typechecking/build, module boundaries, four fresh PostgreSQL +migrations through 0401 and the current-master migration-order gate pass. +The repeated 68 cases are not added to the earlier 304-case matrix; its full +GitHub group remains evidence at its stated revision. + +The terminal cases cover finalized-history quiet, deleted Board sources, pending +presentation and passive results, outer rollback/commit, paused/new runs, +restart/drain/standby, native identity, lease waits, unrelated full-page history, +uncertain publication ownership and tree/workspace/exhausted-recovery producers. +The six new upstream tests cover bounded repeated reads and preserve saved-comment +JSON, identity, scope, author and cancellation semantics in the execution blocker. + +A populated 0400 database upgrades to 0401 with all 399 prior migration receipts +identical and exactly one new receipt. Replay adds none. The terminal index is +valid and ready in both fresh and upgraded databases. The original SQL index +definition is unchanged apart from generation whitespace and filename references. +All prior SQL/journal entries and retained snapshot bytes remain preserved; +canonical retention keeps snapshots 0397–0401. Receipt: +`/tmp/sleep35-upgrade-replay-receipt.json`. + +Actual commands, revision and fixtures: +`/tmp/sleep35-landed-acceptance-command.json`. Logs/receipts: +`/tmp/sleep35-landed-{db-build,types,build,boundaries,github,queued,completion,tree,startup}.*`, +`/tmp/sleep35-landed-migrate-{github,queued,completion,tree}.*`, +`/tmp/sleep35-landed-migration-order.*`, and `/tmp/sleep35-upgrade-replay.*`. +The following audit-only follow-up leaves all owned source/test/migration hashes +unchanged, in `/tmp/sleep35-landed-tested-source-sha256.json`. +Original producer, quietness, failure-isolation, skipped-suite and presentation +results above retain their original scopes. Fresh exact-head CI and Greptile 5/5 +with no actionable findings remain required. This does not establish a full +local repository result, live GitHub delivery or physical host/database sleep. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 859c3f52bc3..83f9ad531fc 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -552,6 +552,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0398_yielding_random.sql", "0399_ambitious_onslaught.sql", "0400_absent_phantom_reporter.sql", + "0401_yummy_ultimates.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0401_yummy_ultimates.sql b/packages/db/src/migrations/0401_yummy_ultimates.sql new file mode 100644 index 00000000000..8af75bf559c --- /dev/null +++ b/packages/db/src/migrations/0401_yummy_ultimates.sql @@ -0,0 +1,3 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_terminal_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_response_comment' + and "chat_actions"."status" = 'processed' and "chat_actions"."delivery_id" is not null + and coalesce("chat_actions"."result"->>'final', 'false') <> 'true'; diff --git a/packages/db/src/migrations/meta/0396_snapshot.json b/packages/db/src/migrations/meta/0401_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0396_snapshot.json rename to packages/db/src/migrations/meta/0401_snapshot.json index 477f53d334e..95b8cabb6fc 100644 --- a/packages/db/src/migrations/meta/0396_snapshot.json +++ b/packages/db/src/migrations/meta/0401_snapshot.json @@ -1,6 +1,6 @@ { - "id": "be87a5d8-4164-4d95-89ae-b895701fa21b", - "prevId": "66fe02b3-6672-4c19-b447-57acad43bf16", + "id": "c4c8a780-68b0-4b79-8c3b-5847f506ed7c", + "prevId": "843ab3c7-1378-42f3-b213-401a8d303e3f", "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,38 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_actions_github_terminal_work_idx": { + "name": "chat_actions_github_terminal_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"status\" = 'processed' and \"chat_actions\".\"delivery_id\" is not null\n and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'", + "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 +12147,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": [ @@ -14996,6 +15106,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": { @@ -30494,6 +30626,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 a434b53ace1..04b623fa766 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2794,6 +2794,13 @@ "when": 1793218595176, "tag": "0400_absent_phantom_reporter", "breakpoints": true + }, + { + "idx": 401, + "version": "7", + "when": 1793219893186, + "tag": "0401_yummy_ultimates", + "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 0e0f56abf4b..e93dafe0b80 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -685,6 +685,9 @@ export const chatActions = pgTable( table.endpointId, table.providerActionId, ), + index("chat_actions_github_terminal_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment' + and ${table.status} = 'processed' and ${table.deliveryId} is not null + and coalesce(${table.result}->>'final', 'false') <> 'true'`), 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'))`), diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index c282a0dfd7f..8a355a4fe8f 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -97,7 +97,8 @@ describe("createChatReconciliationCoordinator", () => { 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(github).not.toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()"); + expect(source).toContain("chatChannels.registerGitHubTerminalResponseWork(deliveryWork,"); 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 d5dd8041ffb..121e55761bb 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js"; +import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js"; import { recordRunPresentationDecision } from "../services/heartbeat/run-completion.js"; import { environmentService } from "../services/environments.js"; import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js"; @@ -4036,6 +4038,392 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { expect(attempts).toBe(2); expect(f.responseWrites).toHaveLength(writes); }); + it("restores terminal GitHub responses and leaves finalized history quiet", async () => { + const f = await toolReplyFixture(); + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), + { timeout: 5000 }).toEqual({ next: null, active: 0 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + const providerWrites = f.responseWrites.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + expect(f.responseWrites).toHaveLength(providerWrites); + } finally { reads.mockRestore(); writes.mockRestore(); } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("settles a deleted Board-origin request without starving unrelated terminal GitHub replies", async () => { + const invalid = await toolReplyFixture(); + const valid = await toolReplyFixture(); + const [message] = await db.insert(issueComments).values({ companyId: invalid.companyId, + issueId: invalid.conversation.issueId!, authorType: "user", authorUserId: "owner-user", body: "Board follow-up", + }).returning(); + await initializeRunIdentity(db, { companyId: invalid.companyId, runId: invalid.run.id, + issueId: invalid.conversation.issueId!, responsibleUserId: "owner-user", messageIds: [message.id], cause: "instruction" }); + await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, message.id)); + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }) + .where(inArray(heartbeatRuns.id, [invalid.run.id, valid.run.id])); + const invalidWrites = invalid.responseWrites.length; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + // The denied request must never reach transport; only the valid request + // uses this provider, even though discovery covers both companies. + const responses = githubResponseCommentService(db, valid.providerFetch); + try { + await responses.registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => [...valid.responseComments.values()][0]?.body, { timeout: 5000 }) + .toContain("stopped before completing"); + const [denied] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, invalid.endpoint.id), + eq(chatActions.kind, "github_response_comment"))); + expect(denied).toMatchObject({ status: "cancelled", result: { retryable: false, code: "authorization_changed" } }); + expect(invalid.responseWrites).toHaveLength(invalidWrites); + expect(valid.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]); + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), + { timeout: 5000 }).toEqual({ next: null, active: 0 }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + 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(); + expect(valid.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]); + } finally { reads.mockRestore(); writes.mockRestore(); } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["passive", "missing"] as const)("keeps native presentation idle until its committed GitHub outcome: %s", async outcome => { + const f = await toolReplyFixture("github:paperclipai/paperclip:418"); + await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "observed", resultJson: null }) + .where(eq(heartbeatRuns.id, f.run.id)); + 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"); + const baseline = f.responseWrites.length, originalBody = [...f.responseComments.values()][0].body; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active })) + .toEqual({ next: null, active: 0 }); + expect(f.responseWrites).toHaveLength(baseline); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + 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(); + expect(f.responseWrites).toHaveLength(baseline); + } finally { reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); } + const settlePresentation = async (tx: Parameters[0]>[0]) => { + await notifyChatRunMilestones(tx, { companyId: f.companyId, issueId: f.conversation.issueId! }); + await tx.update(heartbeatRuns).set({ nativePhase: "committed", updatedAt: new Date(), resultJson: { + finalizationPhase: "committed", finalizationReasonCode: outcome === "passive" ? "external_chat_response_waiting" : "completed", + nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: outcome === "passive" ? "yielded" : "completed", + continuation: outcome === "passive" ? { kind: "response_wake", idempotencyKey: "settled-presentation" } : null }, + presentationDecision: { schema: "paperclip.run_presentation_decision.v1", chosenSource: "none", commentAction: "none", + reasonCodes: [outcome === "passive" ? "yielded_control_plane_wait" : "no_final_response"] }, + } }).where(eq(heartbeatRuns.id, f.run.id)); + }; + await expect(db.transaction(async tx => { await settlePresentation(tx); throw new Error("rollback presentation"); })) + .rejects.toThrow("rollback presentation"); + expect(f.responseWrites).toHaveLength(baseline); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].nativePhase).toBe("observed"); + await db.transaction(settlePresentation); + await expect.poll(async () => (await readChatActionsById(action.id))[0], { timeout: 5000 }).toMatchObject({ + status: outcome === "passive" ? "cancelled" : "processed", result: { final: true }, + }); + if (outcome === "passive") { + expect(f.responseWrites).toHaveLength(baseline); + expect([...f.responseComments.values()][0].body).toBe(originalBody); + } else expect([...f.responseComments.values()][0].body).toContain("without publishing a final response"); + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), { timeout: 5000 }) + .toEqual({ next: null, active: 0 }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("wakes terminal GitHub responses after the outer issue transaction commits", async () => { + const f = await toolReplyFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true); + try { + await worker.ready; + const baseline = f.responseWrites.length; + const change = async (tx: Parameters[0]>[0]) => { + await tx.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(tx as unknown as Db).update(f.conversation.issueId!, { status: "blocked" }); + }; + await expect(db.transaction(async tx => { await change(tx); throw new Error("rollback terminal issue"); })) + .rejects.toThrow("rollback terminal issue"); + expect(f.responseWrites).toHaveLength(baseline); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + const pending = db.transaction(async tx => { await change(tx); entered(); await hold; }); + try { await ready; expect(f.responseWrites).toHaveLength(baseline); } + finally { release(); await pending; } + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("keeps an acknowledged paused GitHub request quiet until its new run finishes", async () => { + const f = await toolReplyFixture("github:paperclipai/paperclip:418"); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id)); + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.conversation.issueId!)); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: f.conversation.externalThreadId }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41802", text: "New request", userId: "42", userName: "octocat", mentioned: true }) }); + const responses = githubResponseCommentService(db, f.providerFetch); + await responses.processPending(); + expect([...f.responseComments.values()][0].body).toContain("paused"); + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await responses.registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + const providerWrites = f.responseWrites.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + expect(f.responseWrites).toHaveLength(providerWrites); + reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); + await coordinator.stop(); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.conversation.issueId!)); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId, + status: "failed", runtimeMode: "native", contextSnapshot: await chatWakeContext({ + endpointId: f.endpoint.id, issueId: f.conversation.issueId, provider: "github", providerMessageId: "41802", + }) }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await responses.registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["restart", "drain", "standby"] as const)("retains terminal GitHub response work across %s", async mode => { + const f = await toolReplyFixture(); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + const responses = githubResponseCommentService(db, f.providerFetch); + let worker = responses.registerTerminalResponseWork(coordinator, () => canRun); + try { + await worker.ready; + const baseline = f.responseWrites.length; + if (mode === "restart") await coordinator.stop(); else canRun = false; + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(db).update(f.conversation.issueId!, { status: "blocked" }); + expect(f.responseWrites).toHaveLength(baseline); + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = responses.registerTerminalResponseWork(coordinator, () => true); + await worker.ready; + } else { canRun = true; worker.wake(); } + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("reactivates terminal GitHub response recovery from committed native identity", async () => { + const f = await toolReplyFixture(); + const messageId = String(f.run.contextSnapshot?.wakeCommentId); + await db.update(heartbeatRuns).set({ status: "failed", nativeIssueId: f.conversation.issueId, + contextSnapshot: {}, updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + const baseline = f.responseWrites.length; + await expect(db.transaction(async tx => { + await initializeRunIdentity(tx as unknown as Db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", messageIds: [messageId], cause: "instruction" }); + throw new Error("rollback identity"); + })).rejects.toThrow("rollback identity"); + expect(f.responseWrites).toHaveLength(baseline); + await initializeRunIdentity(db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", messageIds: [messageId], cause: "instruction" }); + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("waits for the existing GitHub publication lease before a terminal response", async () => { + const f = await toolReplyFixture(); + const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + let lease: Promise | undefined; + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id, + repositoryId: String(resource.metadata.providerRepositoryId), number: 418 }, f.providerFetch, async () => { entered(); await hold; }); + await ready; + const baseline = f.responseWrites.length; + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(db).update(f.conversation.issueId!, { status: "blocked" }); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).not.toBeNull(); + expect(f.responseWrites).toHaveLength(baseline); + release(); await lease; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { release(); await lease; await coordinator.stop(); } + }); + + it("finds owed terminal GitHub work beyond a full page of unrelated request history", async () => { + const f = await toolReplyFixture("github:paperclipai/paperclip:418"); + const [original] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + const [delivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, original.deliveryId!)); + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + const history = Array.from({ length: 101 }, (_, index) => ({ + actionId: `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`, + deliveryId: randomUUID(), commentId: randomUUID(), providerId: `retained-request-${index}`, + })); + await db.insert(issueComments).values(history.map(row => ({ id: row.commentId, companyId: f.companyId, + issueId: f.conversation.issueId!, authorUserId: "owner-user", body: "A separate retained request" }))); + await db.insert(chatDeliveries).values(history.map(row => ({ ...delivery, id: row.deliveryId, + providerEventId: row.providerId, deduplicationKey: row.providerId, state: "processed" as const }))); + await db.insert(chatMessageLinks).values(history.map(row => ({ companyId: f.companyId, endpointId: f.endpoint.id, + conversationId: f.conversation.id, deliveryId: row.deliveryId, commentId: row.commentId, + providerMessageId: row.providerId, direction: "inbound" }))); + await db.insert(chatActions).values(history.map(row => ({ ...original, id: row.actionId, deliveryId: row.deliveryId, + providerActionId: `github_response:${row.deliveryId}`, result: { code: "fast_response_superseded" } }))); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const baseline = f.responseWrites.length; + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(f.responseWrites).toHaveLength(baseline + 1); + expect(await db.select().from(chatActions).where(inArray(chatActions.id, history.map(row => row.actionId)))) + .toEqual(expect.arrayContaining(history.map(row => expect.objectContaining({ id: row.actionId, + status: "processed", result: { code: "fast_response_superseded" } })))); + expect(errors).toEqual([]); + } finally { + await coordinator.stop(); + await db.delete(chatActions).where(inArray(chatActions.id, history.map(row => row.actionId))); + } + }); + + it.each([false, true])("parks terminal GitHub responses behind the publication owner (uncertain: %s)", async uncertain => { + const f = await toolReplyFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let release!: () => void, entered!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { entered = resolve; }); + let intercepted = false; + let sending: Promise | undefined; + const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"))))[0]; + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + f.setSupplementalProviderFetch(async input => { + if (intercepted || !String(input).endsWith("/access_tokens")) return undefined; + const publication = await action(); + if (!publication) return undefined; + intercepted = true; + if (uncertain) await db.update(chatActions).set({ result: { replyWriteStarted: true } }).where(eq(chatActions.id, publication.id)); + await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id)); + await issueService(db).update(f.conversation.issueId!, { status: "blocked" }); + entered(); await hold; + return undefined; + }); + const baseline = f.responseWrites.length; + sending = githubChatReviewService(db, f.providerFetch).execute(f.session, "comment", { body: "Authored final response", idempotencyKey: "terminal-owner-race" }); + await ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(f.responseWrites).toHaveLength(baseline); + release(); await sending; + expect((await action()).status).toBe("cancelled"); + if (!uncertain) await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + if (uncertain) expect(f.responseWrites).toHaveLength(baseline); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + 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 { release(); await sending; await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["tree restore", "workspace admission", "exhausted recovery"] as const)("wakes a paused GitHub response from %s", async producer => { + const f = await toolReplyFixture("github:paperclipai/paperclip:418"); + const issueId = f.conversation.issueId!; + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id)); + await db.update(chatActions).set({ result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || '{"final":true}'::jsonb` }) + .where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment"))); + const thread = makeThread({ channelId: "paperclipai/paperclip", id: f.conversation.externalThreadId }).thread; + await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread, + trigger: "mention", message: makeMessage({ id: "41802", text: "New request", userId: "42", userName: "octocat", mentioned: true }) }); + const [link] = await db.select().from(chatMessageLinks).where(and(eq(chatMessageLinks.endpointId, f.endpoint.id), eq(chatMessageLinks.providerMessageId, "41802"))); + const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.deliveryId, link.deliveryId!), eq(chatActions.kind, "github_response_comment"))))[0]; + let mutate: () => Promise; + if (producer === "tree restore") { + await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issueId)); + const tree = issueTreeControlService(db); + const actor = { actorType: "user" as const, actorId: "owner-user", userId: "owner-user" }; + const cancelled = await tree.createHold(f.companyId, issueId, { mode: "cancel", actor }); + await tree.cancelIssueStatusesForHold(f.companyId, issueId, cancelled.hold.id); + const restore = await tree.createHold(f.companyId, issueId, { mode: "restore", actor }); + mutate = () => tree.restoreIssueStatusesForHold(f.companyId, issueId, restore.hold.id, { actor }); + } else if (producer === "workspace admission") { + const settings = instanceSettingsService(db); + const previous = await settings.getExperimental(); + await settings.updateExperimental({ enableIsolatedWorkspaces: true }); + onTestFinished(() => settings.updateExperimental({ enableIsolatedWorkspaces: previous.enableIsolatedWorkspaces })); + const heartbeat = heartbeatService(db); + onTestFinished(() => heartbeat.drainActiveRunExecutions()); + await db.update(issues).set({ status: "todo", workMode: "standard", responsibleUserId: "owner-user", + projectId: null, projectWorkspaceId: null, executionWorkspaceId: null, executionRunId: null, checkoutRunId: null, + executionWorkspaceSettings: { mode: "isolated_workspace", workspaceStrategy: { type: "git_worktree" } }, + }).where(eq(issues.id, issueId)); + await db.update(agents).set({ runtimeConfig: { heartbeat: { wakeOnDemand: true } } }).where(eq(agents.id, f.assignedAgentId)); + mutate = () => heartbeat.wakeup(f.assignedAgentId, { source: "assignment", triggerDetail: "system", + reason: "issue_assigned", payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" } }); + } else { + await db.update(issues).set({ status: "in_progress", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId)); + await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: issueId, + kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: f.assignedAgentId, + cause: "uncertain_external_action", status: "active", evidence: { runId: f.run.id }, fingerprint: randomUUID(), + nextAction: "Preserve the recorded outcome without replay.", + }); + mutate = () => settleUnrecoverableExecutions(db); + } + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect((await action()).result?.pausedVersion).toBeUndefined(); + await mutate(); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].status).toBe("blocked"); + await expect.poll(async () => (await action()).result?.pausedVersion, { timeout: 5000 }).toEqual(expect.any(String)); + expect([...f.responseComments.values()][0].body).toContain("paused"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + 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)); diff --git a/server/src/app.ts b/server/src/app.ts index daab82394e3..2a09aa0303a 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.registerGitHubTerminalResponseWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index e9855e5cc5e..53b439ec4da 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -3504,6 +3504,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { registerInteractionContinuationWork, + registerGitHubTerminalResponseWork: githubResponseCommentService(db, fetchImpl).registerTerminalResponseWork, registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork, registerTerminalConfirmationWork, loadFailedRetryNoticeScopes, diff --git a/server/src/services/chat-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts index 5ae081cda53..01ccc82250e 100644 --- a/server/src/services/chat-channels/github/maintenance.ts +++ b/server/src/services/chat-channels/github/maintenance.ts @@ -1,5 +1,4 @@ import type { Db } from "@paperclipai/db"; -import { githubResponseCommentService } from "../../chat-github-response-comments.js"; import { githubReviewCheckService } from "../../chat-github-checks.js"; export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fetch, options: { @@ -13,7 +12,6 @@ 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).processTerminalResponses(); await githubReviewCheckService(db, fetchImpl).processPending(); } finally { githubMaintenancePending = false; diff --git a/server/src/services/chat-channels/github/response-execution.ts b/server/src/services/chat-channels/github/response-execution.ts new file mode 100644 index 00000000000..cb9ca365a2e --- /dev/null +++ b/server/src/services/chat-channels/github/response-execution.ts @@ -0,0 +1,42 @@ +import type { Db } from "@paperclipai/db"; +import { githubRunReplyState } from "../../chat-run-publications.js"; +import { readGitHubResponseRun } from "./response-run-state.js"; +export { responseRunColumns as githubResponseRunColumns } from "./response-run-state.js"; + +export type GitHubResponseExecutionInput = { + companyId: string; + endpointId: string; + agentId: string; + issueId: string; + deliveryId: string; + sourceMessageId: string | null; + issueStatus: string; + deliveryCreatedAt: Date; +}; + +export async function githubResponseExecution(db: Db, input: GitHubResponseExecutionInput) { + const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, input); + const requestRunning = ownsLatestRequest && latestRun?.status === "running"; + const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status); + const run = terminal && (ownsLatestRequest || input.issueStatus === "blocked") ? latestRun : null; + + const reply = run && ownsLatestRequest ? await githubRunReplyState(db, { + companyId: input.companyId, endpointId: input.endpointId, issueId: input.issueId, runId: run.id, + }) : "none"; + const paused = !!run ? input.issueStatus === "blocked" && !ownsLatestRequest + : input.issueStatus === "blocked" && latestRun?.status !== "running"; + const presentationPending = !!run && ownsLatestRequest && run.nativePresentationPending && reply === "none"; + const passiveResponseWait = !!run && ownsLatestRequest && run.passiveResponseWait && reply === "none"; + const body = presentationPending ? undefined : run ? reply === "confirmed" ? "This request has been handled. See the response in this conversation." + : paused ? "Work is paused. Your request is saved. See the Paperclip task for details." + : run.status === "succeeded" ? "This turn ended without publishing a final response. See the Paperclip task for details." + : "This turn stopped before completing. See the Paperclip task for details." + : paused ? "Work is paused. Your request is saved. See the Paperclip task for details." : undefined; + return { + body, paused, reply, running: requestRunning, presentationPending, passiveResponseWait, + versionAt: run ? new Date(Math.max(run.updatedAt.getTime(), input.deliveryCreatedAt.getTime())) + : paused ? input.deliveryCreatedAt : undefined, + version: JSON.stringify([latestRun?.id ?? null, latestRun?.status ?? null, + latestRun?.updatedAt.toISOString() ?? null, ownsLatestRequest, input.issueStatus === "blocked", reply]), + }; +} diff --git a/server/src/services/chat-channels/github/terminal-response-work.ts b/server/src/services/chat-channels/github/terminal-response-work.ts new file mode 100644 index 00000000000..c43910d550e --- /dev/null +++ b/server/src/services/chat-channels/github/terminal-response-work.ts @@ -0,0 +1,106 @@ +import { and, asc, eq, gt, isNotNull, sql } from "drizzle-orm"; +import { chatActions, chatConversations, chatDeliveries, chatEndpointLeases, chatEndpointResources, + chatEndpoints, chatMessageLinks, heartbeatRuns, issues, 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"; +import { githubResponseExecution } from "./response-execution.js"; +import { HttpError } from "../../../errors.js"; + +type Scope = ChatResponseScope & { endpointId?: string; repositoryId?: string; number?: number }; +type Entry = { id: string; companyId: string; endpointId: string; deliveryId: string; + issueId: string; conversationId: string; repositoryId: string | null; number: number | null; dueAt: number }; + +export function registerGitHubTerminalResponseWork( + 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]*)?$')`; + async function page(scope?: Scope, cursor?: string, actionId?: string) { + return db.select({ + id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId, + deliveryId: chatActions.deliveryId, conversationId: chatConversations.id, issueId: issues.id, + agentId: chatEndpoints.assignedAgentId, issueStatus: issues.status, + deliveryCreatedAt: chatDeliveries.createdAt, repositoryId, number, result: chatActions.result, + sourceMessageId: sql`(select l.comment_id from ${chatMessageLinks} l + where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId} + and l.delivery_id = ${chatActions.deliveryId} and l.direction = 'inbound' limit 1)`, + 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})`, + }).from(chatActions) + .innerJoin(chatConversations, and(eq(chatConversations.id, chatActions.conversationId), eq(chatConversations.companyId, chatActions.companyId))) + .innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId))) + .innerJoin(chatDeliveries, and(eq(chatDeliveries.id, chatActions.deliveryId), eq(chatDeliveries.companyId, chatActions.companyId))) + .innerJoin(issues, and(eq(issues.id, chatConversations.issueId), eq(issues.companyId, chatActions.companyId))) + .leftJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), eq(chatEndpointResources.companyId, chatActions.companyId))) + .where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.status, "processed"), + isNotNull(chatActions.deliveryId), isNotNull(chatEndpoints.assignedAgentId), + sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`, + sql`(${issues.status} = 'blocked' or (select r.status from ${heartbeatRuns} r + where r.company_id = ${chatActions.companyId} and r.agent_id = ${chatEndpoints.assignedAgentId} + and coalesce(r.native_issue_id::text, r.context_snapshot->>'issueId') = ${chatConversations.issueId}::text + order by r.created_at desc, r.id desc limit 1) in ('succeeded', 'failed', 'interrupted', 'timed_out', 'cancelled'))`, + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(issues.id, 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 pending: Entry = { 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: row.leaseExpiresAt ? Date.parse(row.leaseExpiresAt) : 0 }; + try { + const execution = await githubResponseExecution(db, { ...row, agentId: row.agentId!, deliveryId: row.deliveryId! }); + if (execution.body === undefined || execution.reply === "unsettled" || + (execution.paused && row.result?.pausedVersion === execution.version)) return null; + } catch (error) { + if (!(error instanceof HttpError) || error.status !== 403) throw error; + // Keep this request eligible for the acknowledgement owner's authority + // check and cancellation; it must not poison discovery for other requests. + } + return pending; + } + + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubTerminalResponses, + 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) { + if (!canRun() || signal?.aborted) throw signal?.reason ?? new Error("GitHub terminal-response recovery stopped"); + 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 terminal-response 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 ? entry(remaining) : null; + }, + }, canRun); +} diff --git a/server/src/services/chat-github-publication-lease.ts b/server/src/services/chat-github-publication-lease.ts index 46d9e3b862b..994bb8e9174 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.chatGitHubTerminalResponses, JSON.stringify(scope)); 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 fc1e1354f94..2bb0ecbc5c1 100644 --- a/server/src/services/chat-github-response-comments.ts +++ b/server/src/services/chat-github-response-comments.ts @@ -1,4 +1,3 @@ -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, @@ -6,6 +5,8 @@ import { fastResponseRequests, chatPublications, type Db, } from "@paperclipai/db"; +import { registerGitHubTerminalResponseWork } from "./chat-channels/github/terminal-response-work.js"; +import { githubResponseExecution, githubResponseRunColumns as responseRunColumns } from "./chat-channels/github/response-execution.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { registerGitHubAcknowledgementWork } from "./chat-channels/github/acknowledgement-work.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; @@ -78,7 +79,7 @@ async function stageResponse(db: Db, source: Source) { export async function writeGitHubResponseComment( db: Db, source: Source, api: Api, lease: Lease, - input: { body: string; final: boolean; versionAt?: Date; acknowledgement?: boolean; preserveFinal?: boolean }, + input: { body: string; final: boolean; versionAt?: Date; acknowledgement?: boolean; preserveFinal?: boolean; pausedVersion?: string }, assertCurrent: () => Promise, ) { await assertCurrent(); @@ -101,8 +102,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() }) + if (["received", "failed"].includes(response.status) || (response.status === "processed" && + input.pausedVersion && response.result.pausedVersion !== input.pausedVersion)) await lease.commit(async tx => { + await tx.update(chatActions).set({ status: "processed", + result: { ...response.result, ...(input.pausedVersion ? { pausedVersion: input.pausedVersion } : {}) }, 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) }; @@ -147,6 +150,7 @@ export async function writeGitHubResponseComment( }, updatedAt: new Date() }).where(eq(chatActions.id, target.id)); await tx.update(chatActions).set({ status: "processed", result: { ...receipt, final: input.final || alreadyFinal, versionAt: versionAt.toISOString(), + ...(input.pausedVersion ? { pausedVersion: input.pausedVersion } : {}), }, updatedAt: new Date() }).where(eq(chatActions.id, response.id)); await logActivity(tx as unknown as Db, { companyId: source.endpoint.companyId, actorType: "system", actorId: "github-response", action: "chat_github.response_updated", @@ -207,52 +211,41 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { let requestRunning = false; let requestPaused = false; let preserveFinal = false; + let pausedVersion: string | undefined; if (finalBody === undefined) { const existing = await stageResponse(db, source); - const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, { + const execution = await githubResponseExecution(db, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - agentId: source.endpoint.assignedAgentId, issueId: source.conversation.issueId, - deliveryId, sourceMessageId: source.sourceMessageId, + agentId: source.endpoint.assignedAgentId!, issueId: source.conversation.issueId!, deliveryId, + sourceMessageId: source.sourceMessageId, issueStatus: source.issueStatus, + deliveryCreatedAt: source.delivery.createdAt, }); - 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; - if (run) { - const reply = ownsLatestRequest ? await githubRunReplyState(db, { - companyId: source.endpoint.companyId, endpointId: source.endpoint.id, - issueId: source.conversation.issueId!, runId: run.id }) : "none"; - if (reply === "unsettled") throw conflict("GitHub reply is still being resolved"); - // A confirmed reply for this exact run can retire its placeholder. - // Otherwise terminal execution is not yet a missing-response decision. - if (ownsLatestRequest && run.nativePresentationPending && reply === "none") return; - if (isMainPullResponse(source) && ownsLatestRequest && run.passiveResponseWait && reply === "none") { - // A committed passive turn intentionally has no new response. - // Settle this request's acknowledgement without editing the shared - // PR assessment or inventing a current-run publication receipt. - await lease.commit(async tx => { await tx.update(chatActions).set({ - status: "cancelled", result: { ...existing.result, final: true, code: "passive_response_wait" }, - updatedAt: new Date(), + if (execution.reply === "unsettled") throw conflict("GitHub reply is still being resolved"); + if (execution.presentationPending) return; + if (isMainPullResponse(source) && execution.passiveResponseWait) { + // Preserve the shared assessment when this exact run intentionally waits + // without publishing. Retire only its request acknowledgement. + await lease.commit(async tx => { + await notifyChatResponseAvailability(tx, { companyId: source.endpoint.companyId, deliveryId }); + await tx.update(chatActions).set({ status: "cancelled", + result: { ...existing.result, final: true, code: "passive_response_wait" }, updatedAt: new Date(), }).where(and(eq(chatActions.id, existing.id), eq(chatActions.companyId, source.endpoint.companyId), - sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); }); - return; - } - if (existing.result?.final === true) return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined; - preserveFinal = reply === "confirmed"; - onlyExisting = reply === "confirmed" && !existing.result?.id; - requestPaused = source.issueStatus === "blocked" && !ownsLatestRequest; - finalBody = reply === "confirmed" ? "This request has been handled. See the response in this conversation." - : requestPaused ? "Work is paused. Your request is saved. See the Paperclip task for details." - : run.status === "succeeded" ? "This turn ended without publishing a final response. See the Paperclip task for details." - : "This turn stopped before completing. See the Paperclip task for details."; - finalVersionAt = new Date(Math.max(run.updatedAt.getTime(), source.delivery.createdAt.getTime())); - } else if (source.issueStatus === "blocked" && latestRun?.status !== "running") { - requestPaused = true; - finalBody = "Work is paused. Your request is saved. See the Paperclip task for details."; - finalVersionAt = source.delivery.createdAt; - } else if (existing.result?.id && (!isMainPullResponse(source) || + sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); + }); + return; + } + if (existing.result?.final === true) return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined; + if (execution.paused && existing.status === "processed" && existing.result?.id && existing.result?.pausedVersion === execution.version) + return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined; + pausedVersion = execution.paused ? execution.version : undefined; + requestRunning = execution.running; + requestPaused = execution.paused; + preserveFinal = execution.reply === "confirmed"; + onlyExisting = preserveFinal && !existing.result?.id; + finalBody = execution.body; + finalVersionAt = execution.versionAt; + if (finalBody === undefined && existing.result?.id && (!isMainPullResponse(source) || (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({ status: ["received", "failed"].includes(existing.status) ? "processed" : existing.status, updatedAt: new Date(), @@ -293,7 +286,7 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { } return writeGitHubResponseComment(db, source, api, lease, { body: finalBody ?? (isMainPullResponse(source) && !requestRunning ? "Review queued…" : "Working on this…"), final: finalBody !== undefined && !requestPaused, - acknowledgement: finalBody === undefined, preserveFinal, + acknowledgement: finalBody === undefined, preserveFinal, pausedVersion, ...(finalBody !== undefined ? { versionAt: finalVersionAt ?? new Date() } : {}) }, assertCurrent); }); } @@ -454,6 +447,8 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) { } } return { tryAcknowledge, processPending, processTerminalResponses, failRun, finishRun, hasWorkingResponseForRun, + registerTerminalResponseWork: (coordinator: ReturnType, canRun: () => boolean) => + registerGitHubTerminalResponseWork(db, coordinator, tryAcknowledge, canRun), registerAcknowledgementWork: (coordinator: ReturnType, canRun: () => boolean) => registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun), }; diff --git a/server/src/services/chat-github-reviews.ts b/server/src/services/chat-github-reviews.ts index 5fb2be6d1a6..06102e8df4b 100644 --- a/server/src/services/chat-github-reviews.ts +++ b/server/src/services/chat-github-reviews.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { githubRunReplyState } from "./chat-run-publications.js"; import { stageReceiptReactionRemovals } from "./chat-receipt-reactions.js"; import { chatRunSourceMessageId } from "./chat-channels/run-origin.js"; @@ -461,6 +462,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { ): Promise { if (writer === db) return db.transaction(tx => stage(source, operation, payload, key, invocationId, toolName, parameters, tx)); const actionKey = `github_publication:${source.issue.id}:${operation}:${hash(key)}`; + await notifyChatResponseAvailability(writer, { companyId: source.endpoint.companyId, issueId: source.issue.id }); await notifyDeliveryWork(writer, DELIVERY_QUEUES.chatGitHubPublications); const values = { companyId: source.endpoint.companyId, @@ -1046,6 +1048,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { )?.[1]; if (!number || !binding?.repository.providerRepositoryId) { await db.transaction(async tx => { + await notifyChatResponseAvailability(tx, { companyId: action.companyId, conversationId: action.conversationId ?? undefined }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); await tx.update(chatActions).set({ status: "cancelled", result: { code: "publication_binding_unavailable", retryable: false }, updatedAt: new Date() }) .where(and(eq(chatActions.id, actionId), eq(chatActions.companyId, action.companyId))); diff --git a/server/src/services/chat-response-availability.ts b/server/src/services/chat-response-availability.ts index 29de7ca303e..d1459d65d9d 100644 --- a/server/src/services/chat-response-availability.ts +++ b/server/src/services/chat-response-availability.ts @@ -3,5 +3,6 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string }; export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); 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 1727d2eeca3..b277485b7e2 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -9,6 +9,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type ChatRunMilestoneScope = { companyId: string; issueId?: string }; export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); // Lifecycle and coarse native progress share lineage and authority. Extend // this eligibility boundary once rather than adding wrappers to its callers. diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index cab6f3b8c94..20bb0a4b808 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", + chatGitHubTerminalResponses: "chat-github-terminal-responses", chatGitHubAcknowledgements: "chat-github-acknowledgements", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", diff --git a/server/src/services/execution-recovery-resolution.ts b/server/src/services/execution-recovery-resolution.ts index 59b2b12cf5a..e30808f1025 100644 --- a/server/src/services/execution-recovery-resolution.ts +++ b/server/src/services/execution-recovery-resolution.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { notifyExecutionRunWork } from "./execution-control-notifications.js"; import { hasWorkspaceRestoreFailure } from "@paperclipai/shared"; @@ -587,6 +588,7 @@ export async function settleUnrecoverableExecutions( let nativeFailureBlock = action.evidence.nativeFailureBlock; let nativeBootstrapFailureBlock = action.evidence.nativeBootstrapFailureBlock; if (current) { + if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id }); const [projected] = await tx .update(issues) .set({ diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 580b46e298c..bd1694c3a41 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "../chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { assertTaskRetryConversationAllowed, readTaskRetryIntent, readTaskRetryBoardMessageOrigin, retireSupersededTaskRetry, taskRetryMutationAllowed } from "../task-retry.js"; @@ -3863,6 +3864,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende `- Reason: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_MESSAGE}`, `- Next action: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION}`, ].join("\n"); + if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id }); await tx .update(issues) .set({ diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 227fae5948f..0ac70e83077 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -999,7 +1000,10 @@ export function issueTreeControlService(db: Db) { const restored: TreeStatusUpdateResult["updatedIssues"] = []; for (const [status, issueIdsForStatus] of issueIdsByStatus) { if (issueIdsForStatus.length === 0) continue; - for (const issueId of issueIdsForStatus) await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); + for (const issueId of issueIdsForStatus) { + await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`); + await notifyChatResponseAvailability(tx, { companyId, issueId }); + } const rows = await tx .update(issues) .set({ diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 8185a4881c5..25bf82319a2 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,3 +1,4 @@ +import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyApprovalWork } from "./approval-work-notifications.js"; @@ -11430,6 +11431,10 @@ export function issueService(db: Db) { if (patch.status === "done" || patch.status === "cancelled") { await notifyDecisionTargetWork(tx, `issue:${id}`); } + if ((patch.status !== undefined && patch.status !== receiptExisting.status) || + (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId)) { + await notifyChatResponseAvailability(tx, { companyId: receiptExisting.companyId, issueId: id }); + } if (patch.status !== undefined && patch.status !== receiptExisting.status) { await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId, issueId: id }); }