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