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 10cc1ee7681..992cca11f32 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2398,3 +2398,43 @@ 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. + +## Slice 35 accepted; slice 36 wired + +Slice 35 is [#17661](https://github.com/paperclipai/paperclip/pull/17661), head +`3d7c79c262f92563555b278d101de8240e6f011b`. Complete local chat/publication/startup +acceptance passed all 1,571 tests in 517.41 seconds. All 53 remote checks passed, +and fresh Greptile is 5/5 with no findings. The PR remains draft and unmerged. + +Slice 36, `codex/sleep-36-github-review-checks`, owns GitHub review-check publication. +It uses existing actions and scoped scheduling. Received nonterminal checks retain +the existing five-second remote-head reconciliation deadline because webhook +receipt is not a reliable proof of all future pushes. Completed history and +configuration alone do not schedule work. Current authority, assessment ownership, +provider identity, check replacement, lease fencing, and failure backoff remain +with the GitHub owner. Late stale-head projection repair is the separate next duty. + +Eleven new worker tests cover real enqueue and nested transaction settlement, +run-identity wake before a saved deadline, missed-head recovery, lifecycle states, +retry reconstruction, lease contention, a full revoked prefix, retained history, +and HTTP shutdown recovery. The fresh targeted selection passed 36 tests including +existing maintenance/startup cases. Workspace typecheck/build, migration/replay, +and candidate-index evidence passed. Full local acceptance and remote PR gates +remain required. See `2026-10-25-github-review-check-sleep-audit.md` for the failed +attempts, repairs, exact boundary, and external reconciliation limitation. + +Slice 36 is draft [#17662](https://github.com/paperclipai/paperclip/pull/17662). +Its initial full local run had 13 failures and 1,569 passes, and CI failed an +inherited login-cleanup fixture. The quiet observation and fake-clock settlement +repairs are documented in its owner audit. The corrected selections passed 23 +chat cases and all 16 login-cleanup tests; full/new-head verification is required. +The earlier cancellation browser failures were not reproduced in six local +replays. #17490 and #17572 remain unresolved gates, not inferred green from that +negative result. The next untouched duty remains late stale-head projection repair. + +Slice 36 follow-up: all 53 remote checks and fresh 5/5 review passed on +`59f8c8b3e5`. The second local full run had one startup-quiescence assertion +failure and 1,581 passes. Waiting for the actual quiet condition fixes that +fixture; all 211 GitHub workflow tests pass after the change. Fresh remote gates +for the fixture-only commit remain required. The next slice may build on this +commit while those independent gates run; no stack merge is authorized. diff --git a/doc/plans/2026-10-25-github-review-check-sleep-audit.md b/doc/plans/2026-10-25-github-review-check-sleep-audit.md new file mode 100644 index 00000000000..ef5eaf314f1 --- /dev/null +++ b/doc/plans/2026-10-25-github-review-check-sleep-audit.md @@ -0,0 +1,211 @@ +# GitHub review-check scheduling audit + +Baseline: `3d7c79c262f92563555b278d101de8240e6f011b`, parent +[#17661](https://github.com/paperclipai/paperclip/pull/17661). +Parent acceptance: 1,571 tests across complete chat, publication, and startup +suites passed in 517.41 seconds. All 53 remote checks passed on that commit; +Greptile reviewed the same head at 5/5 with no findings. The parent is unmerged. + +## Scope and owner + +This slice owns publication of the GitHub Review check derived from ordinary +chat admission, review binding, run state, and assessment ownership. Existing +`chat_actions.kind = github_review_check` rows remain durable truth. No new +queue table is needed. GitHub state interpretation belongs under +`chat-channels/github/`; the shared coordinator receives identifiers/deadlines. + +Late stale-head projection repair is a separate duty. The manual combined +`processPending` API remains compatible. The legacy maintenance callback now +calls only `repairSupersededReviews`; removing that scan is the next slice. +Assessment publication remains owned by its existing governed outbox, including +its authorization, score, delivery receipts, and recovery guarantees. + +## Actual external obligation + +The old dispatcher checks the current remote PR head before publishing. A saved +nonterminal check receives a five-second `retryAt` even when its local state has +not changed. The current webhook parser accepts opened, synchronize, reopened, +and ready-for-review events, but receipt is not a proof that all future pushes +will arrive. A manual review can also exist without a new automatic admission. +This slice must not turn a missed webhook into a permanently active obsolete +check or authorize publication against an old head. + +Retain the existing five-second remote-head reconciliation deadline only while +that particular check is received/nonterminal. Local state changes can make the +entry due immediately; a successful unchanged inspection does not rewrite the +provider check. Failure uses its saved exponential backoff and existing eight- +attempt exhaustion policy. Processed, cancelled, and exhausted history have no +worker deadline. This is not a claim that an active review needs no SQL/network +traffic: unresolved checks retain external reconciliation until they settle. +The startup global scan and committed scope rebuilds enumerate unfinished rows; +deadline dispatch rechecks only known entry IDs. No configuration-only timer. + +## Producer and lifecycle boundary + +- `githubReviewCheckService.enqueue`: automatic webhook and authorized manual + creation use one awaited notification before insertion, inside a database + transaction. Caller-owned/nested transaction settlement stays with `createDb`. +- `retainGitHubReviewReceipt`: task admission signals its repository/PR before + inserting the queued review in the existing caller-owned transaction. +- `reviewForHead`: manual review insertion and binding an existing review to a + run signal inside their short database transactions. Provider calls remain + outside those transactions. +- Existing run/lineage and endpoint-configuration notification ownership fans + out to the new topic by company. These hints conservatively rebuild that + company's outstanding checks; they do not inspect completed history. Narrow + repository/PR hints are used by check enqueue, review writes, and lease release. +- Assessment and publication receipts are serialized by the existing GitHub + publication lease. Its transactional release wakes the affected repository/PR. + A competing holder's saved expiry is a deadline if it dies without release. +- Delivery admission failures retain discovery through the outstanding check's + remote reconciliation deadline. This does not claim every shared delivery + mutation now emits a check-specific immediate signal. +- Restart reconstructs existing received/retryable-failed rows. Shared scheduling + retains dirty scopes across drain/standby. HTTP admission is checked before each + provider request; receipt fencing remains with the publication lease. +- Supported writers are the existing process-local database ownership boundary. + Independent SQL writers and other replicas do not receive a new signal protocol. + +Current provider-state reads preserve the prior latest-review/run selection, +legacy delivery provenance, current configuration checks, and completed-check +replacement behavior. Introducing a scheduler does not authorize changing those +product decisions or passing GitHub payloads into generic scheduling machinery. + +## Evidence ledger + +- Provider-state extraction: server typecheck initially found widened returned + state literals. Explicit domain-state typing repaired the extraction. +- Three existing check/late-projection/manual-review acceptance cases passed after + extraction. This was a targeted selection, not all GitHub coverage. +- First worker selection: 12 passed, one retry test failed. The fixture installed + fake timers after a real 60-second timer had been armed, so fake advancement did + not deliver that timer. Move fake timer installation before registration; keep + the production deadline unchanged. Corrected selection pending. + +Remaining acceptance: immediate producer transitions, scope/history pagination, +provider failure and drain boundaries, startup wiring, migration/index evidence, +full relevant acceptance, workspace checks, and remote PR review/CI. No deployed +host sleep, Cloud alarm coverage expansion, or fleet savings are claimed. + +The corrected first selection passed all 13 cases. Expanded coverage added real +run-identity commit/rollback, a 101-row revoked prefix, and shutdown during HTTP. +It exposed a real stop-admission error: the HTTP boundary wrapped the stop as a +provider error and spent a retry attempt. Check admission in the catch boundary +as well; leave the durable action for resume. That regression then passed. The +next run passed 15 cases and failed the history fixture: fixed UUIDs collided +with intentionally retained rows from earlier tests. Use a per-fixture UUID +prefix while retaining deterministic before-candidate ordering. No database was +dropped and no production timeout was enlarged. Workspace typecheck passed +before the final shutdown guard change; final acceptance remains pending. + +## Wired validation checkpoint + +The fresh database selection passed 36 cases across chat integration and startup +wiring in 63.66 seconds. It includes eleven new worker cases plus selected existing +check and maintenance cases; the other 1,532 cases were filtered out. The complete +chat/publication/startup selection is now running separately in a fresh database. +Workspace typecheck and build passed after the shutdown repair. + +Migration 0392 adds only `chat_actions_github_check_work_idx`, with the exact +received/retryable-failed predicate used by the worker. It uses concurrent, +idempotent creation and the existing top-level migration runner. Generation, +safety checks, application, replay, and index validity passed. Snapshot pruning +retains 0388 through 0392. A rollback-only plan inserted 10,000 processed actions +and one unfinished action; candidate selection used an index-only scan, read one +row and three shared buffers, and took 0.019 ms. This measures candidate selection, +not all provenance reads or fleet-wide performance. + +The earlier source checkpoint is `219061a159`. No full local acceptance or remote +CI success is claimed for this slice until the running checks finish. Open parent +stack failures and review findings remain independent obligations; this slice does +not make the whole stack ready or authorize merging it. + +## Full acceptance and CI repairs + +Draft [#17662](https://github.com/paperclipai/paperclip/pull/17662) was published at +`c0d91f553994ba898edc2701c6879c10a46b8b50`. Its first full local run completed with +1,569 passes and 13 failures in 541.25 seconds. The first failure observed a final +terminal-response state read after the scheduler removed its deadline but before +the in-flight pass finished. `nextWakeAt() === null` does not prove quiescence. +That assertion also skipped restoring its SQL spies, causing cascading failures. +The affected quiet checks now require both zero tracked active work and no next +wake. Their spies also restore on test completion if an assertion fails. All 23 +affected chat cases passed in a fresh database after this repair. + +The same head's CI had 51 successes and the server 1/12 plus aggregate failure. +Server 1/12 failed the inherited setup-token consume fixture while it waited for +an idle hold. Cleanup can signal another generation during a sweep, leaving a +coalesced deadline beyond the fixture's single fake-clock jump. The fixture now +advances to actual registered deadlines while tracked work remains. It keeps the +existing observation timeout and all production intervals. All 16 login-cleanup +cases passed in a separate disposable database. Fresh full acceptance and new- +head CI/review remain required; the old head is not reported green. + +An independent diagnostic replayed the earlier stack's two failing cancellation +browser cases three times each: all six passed on this head in 33.6 seconds. +Scoped lock observation did not capture a deadlock. A second minimal FK/row-lock +hypothesis also did not reproduce a cycle. These are negative diagnostic results, +not a fix for #17490 or #17572. An initial database-name collision stopped setup; +the existing database was preserved and a newly named disposable database was +used instead. No blind CI rerun was consumed. + +The second complete local run finished with 1,581 passes and one failure in +541.48 seconds. The remaining assertion checked no deadline immediately after +`ready`, which awaits the first pass, not a later coalesced wake. Fixture teardown +already retires unfinished rows. Startup quiet now waits for both zero active +work and no deadline, matching the strengthened post-delivery assertion. The +complete GitHub workflow selection then passed all 211 tests in 187.16 seconds +in a new database. No production behavior changed. Head `59f8c8b3e5` has all 53 +remote checks passing and fresh Greptile 5/5 with zero review threads. The new +fixture-only commit requires its own fresh remote checks; the two local full +runs are not reported as passes. + +## Current-parent merge acceptance (October 28) + +The earlier results above retain their stated revisions. They are not evidence +that an old review score applies to a repaired head. + +The existing branch was merged normally with the updated terminal-response +parent. Only its unlanded `0396_charming_edwin_jarvis` migration artifacts were +removed before canonical generation. The domain and schema edits were preserved. +Migration `0402_ancient_malice.sql` now creates the same partial work index with +concurrent, idempotent creation. All parent SQL, journal entries, and retained +snapshot bytes remain unchanged. Generation retains snapshots 0398 through 0402. + +At `6995f84b240a14a7b8660e1cf095b3aa1e1e3168`, 331 distinct cases passed: +264 complete GitHub workflow cases, 30 completion cases, eight real/unit issue-tree +producer cases, 16 login-cleanup cases, and 13 startup cases. The GitHub command +filtered 1,485 unrelated cases. Login cleanup used its real embedded PostgreSQL +fixture; the other database groups used separate, pre-migrated PostgreSQL fixtures. +Database build, server typecheck/build, module boundaries, migration safety, and +target-branch migration order passed. + +The actual terminal-response parent landed as +`f8226b0515634f9c7c74e80bb6cf18f3c2634c12`. Its squash tree exactly matched its +verified merge composition. A normal merge then brought that actual parent and +current master into this branch at +`9c538aed1251fd7f7cecb829a0ecbce6110fb1ed`. Both complete added/removed patches +and every unrelated master blob were preserved. This includes the privacy-query, +remote native skill delivery, lockfile, and upstream Telegram fixture changes. + +At that exact integration, all 264 complete GitHub workflow cases and all +13 startup cases passed again: 277 distinct cases, with 1,485 unrelated +GitHub cases filtered out. A new database received all migrations through 0402. +Database build, server typecheck/build, module boundaries, and migration order +against the actual landed parent passed. The earlier 331-case result retains +its earlier revision scope. + +A populated parent database upgraded from 401 to 402 with all 400 previous +migration receipts unchanged and exactly one new receipt. Replay added none. +The index was valid and ready on both that database and a fresh database. +A rollback-only candidate probe added 10,000 processed actions and one received +action. The actual parameterized predicate selected one row through the partial +index with two shared buffers and 0.007 ms execution time. This measures candidate +discovery, not all provenance or lease reads. The first probe omitted the required +provider action ID and rolled back; the corrected fixture supplied unique IDs. +No production code changed for that fixture error. + +The final audit commit changes documentation only. Source, tests, and migration +hashes match the accepted integration. Fresh current-head CI and Greptile review +remain required before merge. No host suspension, external alarm expansion, or +fleet cost reduction is claimed. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 83f9ad531fc..f9a2c6032a0 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -553,6 +553,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0399_ambitious_onslaught.sql", "0400_absent_phantom_reporter.sql", "0401_yummy_ultimates.sql", + "0402_ancient_malice.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0402_ancient_malice.sql b/packages/db/src/migrations/0402_ancient_malice.sql new file mode 100644 index 00000000000..d2831284744 --- /dev/null +++ b/packages/db/src/migrations/0402_ancient_malice.sql @@ -0,0 +1,2 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_check_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_review_check' + and ("chat_actions"."status" = 'received' or ("chat_actions"."status" = 'failed' and "chat_actions"."result"->>'retryable' = 'true')); \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0397_snapshot.json b/packages/db/src/migrations/meta/0402_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0397_snapshot.json rename to packages/db/src/migrations/meta/0402_snapshot.json index 648ee0ff654..c4d702a1cc5 100644 --- a/packages/db/src/migrations/meta/0397_snapshot.json +++ b/packages/db/src/migrations/meta/0402_snapshot.json @@ -1,6 +1,6 @@ { - "id": "c7947622-f079-44c9-a57a-5a11d1bec5e8", - "prevId": "be87a5d8-4164-4d95-89ae-b895701fa21b", + "id": "6ce76555-38d9-4074-873f-db02c546e2c6", + "prevId": "c4c8a780-68b0-4b79-8c3b-5847f506ed7c", "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,54 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_actions_github_check_work_idx": { + "name": "chat_actions_github_check_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_review_check'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))", + "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 +12163,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": [ @@ -30516,6 +30642,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 04b623fa766..60652c8d108 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2801,6 +2801,13 @@ "when": 1793219893186, "tag": "0401_yummy_ultimates", "breakpoints": true + }, + { + "idx": 402, + "version": "7", + "when": 1793223470886, + "tag": "0402_ancient_malice", + "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 e93dafe0b80..165bf7f86df 100644 --- a/packages/db/src/schema/chat_channels.ts +++ b/packages/db/src/schema/chat_channels.ts @@ -685,6 +685,8 @@ export const chatActions = pgTable( table.endpointId, table.providerActionId, ), + index("chat_actions_github_check_work_idx").on(table.id).where(sql`${table.kind} = 'github_review_check' + and (${table.status} = 'received' or (${table.status} = 'failed' and ${table.result}->>'retryable' = 'true'))`), 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'`), diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 8a355a4fe8f..14812e22085 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -99,6 +99,8 @@ describe("createChatReconciliationCoordinator", () => { expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processPending()"); expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()"); expect(source).toContain("chatChannels.registerGitHubTerminalResponseWork(deliveryWork,"); + expect(source).toContain("chatChannels.registerGitHubReviewCheckWork(deliveryWork,"); + expect(github).not.toContain("githubReviewCheckService(db, fetchImpl).processPending()"); 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 121e55761bb..8c062b12a8f 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -4351,15 +4351,18 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { 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(); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 5000 }).toEqual({ active: 0, next: null }); 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(); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 5000 }).toEqual({ active: 0, next: null }); 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"); + onTestFinished(() => { reads.mockRestore(); writes.mockRestore(); }); await vi.advanceTimersByTimeAsync(120_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); reads.mockRestore(); writes.mockRestore(); expect(errors).toEqual([]); @@ -7249,6 +7252,237 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { runId: run.id, }, "read_pull_request", { section: "metadata" })).rejects.toThrow("older pull request head"); }); + async function reviewCheckWorkerFixture(status: "running" | "failed" = "failed") { + const f = await toolReplyFixture("github:paperclipai/paperclip:418"); + const config = await f.management.configuration(f.endpoint.id, "owner-user"); + const event = githubAutomaticReviewEvent({ action: "opened", repository: { id: 97531, full_name: "paperclipai/paperclip" }, sender: { id: 42, login: "octocat" }, + pull_request: { number: 418, title: "Review", body: "", draft: false, base: { sha: "a".repeat(40), ref: "master" }, head: { sha: "b".repeat(40) }, user: { id: 42, login: "octocat" }, labels: [] } }, randomUUID())!; + const [review] = await db.insert(chatGitHubReviews).values({ companyId: f.companyId, endpointId: f.endpoint.id, + issueId: f.conversation.issueId!, runId: f.run.id, repositoryId: "97531", repository: event.repository, pullNumber: 418, headSha: event.headSha, + deliveryId: event.deliveryId, configurationRevision: config.revision, policySnapshot: config.configuration.defaults, event }).returning(); + await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.run.id)); + const writes: Record[] = [], requests: string[] = []; + let remoteHead = event.headSha, unavailable = false; + f.setSupplementalProviderFetch(async (input, init) => { + const url = String(input); + if (url.endsWith("/pulls/418")) { requests.push(url); return Response.json({ head: { sha: remoteHead } }); } + if (url.includes("/check-runs?")) { requests.push(url); return Response.json({ check_runs: [] }); } + if (url.endsWith("/check-runs") && init?.method === "POST") { + requests.push(url); + if (unavailable) return Response.json({ message: "injected check outage" }, { status: 503 }); + writes.push(JSON.parse(String(init.body))); + return Response.json({ id: 418, html_url: "https://github.com/checks/418" }); + } + return undefined; + }); + const checks = githubReviewCheckService(db, f.providerFetch); + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "github_review_check"))))[0]; + return { ...f, event, review, endpoint, checks, writes, requests, action, + enqueue: () => checks.enqueue(endpoint, event, true, "authorized"), + setRemoteHead: (value: string) => { remoteHead = value; }, setUnavailable: (value: boolean) => { unavailable = value; } }; + } + + it("keeps GitHub review check history quiet and wakes from the real enqueue producer", async () => { + const f = await reviewCheckWorkerFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await f.checks.registerCheckWork(coordinator, () => true).ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 5000 }).toEqual({ active: 0, next: null }); + await f.enqueue(); + await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed"); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 5000 }).toEqual({ active: 0, next: null }); + expect(f.writes).toEqual([expect.objectContaining({ status: "completed" })]); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"); + onTestFinished(() => { reads.mockRestore(); transactions.mockRestore(); }); + const count = f.requests.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); + expect(f.requests).toHaveLength(count); + reads.mockRestore(); transactions.mockRestore(); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("defers GitHub review check enqueue until outer commit and discards rollback", async () => { + const f = await reviewCheckWorkerFixture(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const enqueue = (tx: Parameters[0]>[0]) => + githubReviewCheckService(tx as unknown as Db, f.providerFetch).enqueue(f.endpoint, f.event, true, "authorized"); + try { + await f.checks.registerCheckWork(coordinator, () => true).ready; + await expect(db.transaction(async tx => { await enqueue(tx); throw new Error("rollback check"); })).rejects.toThrow("rollback check"); + expect(await f.action()).toBeUndefined(); expect(f.writes).toEqual([]); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }), hold = new Promise(resolve => { release = resolve; }); + const pending = db.transaction(async tx => { await enqueue(tx); entered(); await hold; }); + try { await ready; expect(f.writes).toEqual([]); } finally { release(); await pending; } + await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed"); + expect(f.writes).toHaveLength(1); + } finally { await coordinator.stop(); } + }); + + it("recovers a missed GitHub head webhook only while a review check remains unresolved", async () => { + const f = await reviewCheckWorkerFixture("running"); + await f.enqueue(); + const errors: unknown[] = []; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await f.checks.registerCheckWork(coordinator, () => true).ready; + await expect.poll(() => f.writes.length, { timeout: 5000 }).toBe(1); + expect(f.writes[0].status).toBe("in_progress"); + const saved = await f.action(); + expect(Date.parse(String(saved.result?.retryAt))).toBeGreaterThan(Date.now()); + f.setRemoteHead("c".repeat(40)); + await vi.advanceTimersByTimeAsync(6000); + await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("cancelled"); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 5000 }).toEqual({ active: 0, next: null }); + expect((await db.select().from(chatGitHubReviews).where(eq(chatGitHubReviews.id, f.review.id)))[0].state).toBe("superseded"); + const reads = vi.spyOn(db, "select"), count = f.requests.length; + onTestFinished(() => reads.mockRestore()); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(f.requests).toHaveLength(count); reads.mockRestore(); + expect(f.writes).toHaveLength(1); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each(["restart", "drain", "standby"] as const)("retains GitHub review check work across %s", async mode => { + const f = await reviewCheckWorkerFixture(); + let canRun = true; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + let worker = f.checks.registerCheckWork(coordinator, () => canRun); + try { + await worker.ready; + if (mode === "restart") await coordinator.stop(); else canRun = false; + await f.enqueue(); expect(f.writes).toEqual([]); + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = f.checks.registerCheckWork(coordinator, () => true); await worker.ready; + } else { canRun = true; worker.wake(); } + await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); + expect(f.writes).toHaveLength(1); expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("retains the saved GitHub review check retry through restart without an early provider call", async () => { + const f = await reviewCheckWorkerFixture(); + f.setUnavailable(true); await f.enqueue(); await f.checks.processPending(); + const failed = await f.action(); + expect(failed).toMatchObject({ status: "failed", result: { attempts: 1, retryable: true } }); + const future = new Date(Date.now() + 60_000).toISOString(); + await db.update(chatActions).set({ result: { ...failed.result, retryAt: future } }).where(eq(chatActions.id, failed.id)); + f.setUnavailable(false); + const errors: unknown[] = []; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const count = f.requests.length; + await f.checks.registerCheckWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBe(Date.parse(future)); expect(f.requests).toHaveLength(count); + await vi.advanceTimersByTimeAsync(61_000); + await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("processed"); + expect(f.writes).toHaveLength(1); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("waits for and wakes from the actual GitHub review check publication lease", async () => { + const f = await reviewCheckWorkerFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }), hold = new Promise(resolve => { release = resolve; }); + let lease: Promise | undefined; + try { + await f.checks.registerCheckWork(coordinator, () => true).ready; + lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id, repositoryId: "97531", number: 418 }, f.providerFetch, + async () => { entered(); await hold; }); + await ready; await f.enqueue(); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).not.toBeNull(); + expect(f.writes).toEqual([]); release(); await lease; + await expect.poll(async () => (await f.action())?.status, { timeout: 5000 }).toBe("processed"); + expect(f.writes).toHaveLength(1); expect(errors).toEqual([]); + } finally { release(); await lease; await coordinator.stop(); } + }); + + it("wakes a GitHub review check from committed run identity before its future head deadline", async () => { + const f = await reviewCheckWorkerFixture("running"); + await f.enqueue(); await f.checks.processPending(); + const action = await f.action(), future = new Date(Date.now() + 60_000).toISOString(); + await db.update(chatActions).set({ result: { ...action.result, retryAt: future } }).where(eq(chatActions.id, action.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await f.checks.registerCheckWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBe(Date.parse(future)); + 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 initializeRunIdentity(tx as unknown as Db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, + responsibleUserId: "owner-user", messageIds: [String(f.run.contextSnapshot?.wakeCommentId)], cause: "instruction" }); + }; + await expect(db.transaction(async tx => { await change(tx); throw new Error("rollback run check"); })).rejects.toThrow("rollback run check"); + expect(f.writes).toHaveLength(1); + await db.transaction(change); + await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("processed"); + expect(f.writes).toHaveLength(2); expect(f.writes[1].status).toBe("completed"); + expect(Date.now()).toBeLessThan(Date.parse(future)); expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("finds a GitHub review check after a full page of revoked work and ignores settled history", async () => { + const f = await reviewCheckWorkerFixture(); + await f.enqueue(); const original = await f.action(); + const pageKey = randomUUID().replaceAll("-", "").slice(0, 9); + const prefix = Array.from({ length: 101 }, (_, index) => ({ ...original, + id: `00000000-0000-4000-8000-${pageKey}${String(index + 1).padStart(3, "0")}`, + providerActionId: `revoked-check-${index}`, payload: { ...original.payload, runtimeGeneration: -1 } })); + await db.insert(chatActions).values(prefix); + await db.insert(chatActions).values(Array.from({ length: 101 }, (_, index) => ({ ...original, + id: randomUUID(), providerActionId: `settled-check-${index}`, status: "processed" }))); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await f.checks.registerCheckWork(coordinator, () => true).ready; + await expect.poll(async () => (await db.select().from(chatActions).where(and(eq(chatActions.id, original.id))))[0].status, + { timeout: 15_000 }).toBe("processed"); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 15_000 }).toBeNull(); + expect(f.writes).toHaveLength(1); + const cancelled = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.status, "cancelled"))); + expect(cancelled).toHaveLength(101); expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("stops GitHub review check HTTP after shutdown admission and recovers its durable action", async () => { + const f = await reviewCheckWorkerFixture(); + await f.enqueue(); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }), hold = new Promise(resolve => { release = resolve; }); + const guardedProvider: typeof fetch = async (input, init) => { + if (String(input).endsWith("/pulls/418")) { entered(); await hold; } + return f.providerFetch(input, init); + }; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const starting = githubReviewCheckService(db, guardedProvider).registerCheckWork(coordinator, () => true).ready; + await ready; + const stopped = coordinator.stop(); release(); await stopped; await starting; + expect(f.writes).toEqual([]); expect((await f.action()).status).toBe("received"); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await f.checks.registerCheckWork(coordinator, () => true).ready; + await expect.poll(async () => (await f.action()).status, { timeout: 5000 }).toBe("processed"); + expect(f.writes).toHaveLength(1); expect(errors).toEqual([]); + } finally { release(); await coordinator.stop(); } + }); + it("supersedes a late review projection after its check already rejected the old head", async () => { const f = await reviewBotFixture(); const deliveryId = randomUUID(); diff --git a/server/src/app.ts b/server/src/app.ts index 2a09aa0303a..abbfc3b469f 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.registerGitHubReviewCheckWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerGitHubTerminalResponseWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 53b439ec4da..d4ec773f63f 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,4 @@ +import { githubReviewCheckService } from "./chat-github-checks.js"; import { leasedChatProvider } from "./chat-channels/runtime/receiver-policy.js"; import { chatReceiverSchedule } from "./chat-channels/runtime/receiver-schedule.js"; import { registerChatReceiverWork } from "./chat-channels/runtime/receiver-work.js"; @@ -3504,6 +3505,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { registerInteractionContinuationWork, + registerGitHubReviewCheckWork: githubReviewCheckService(db, fetchImpl).registerCheckWork, registerGitHubTerminalResponseWork: githubResponseCommentService(db, fetchImpl).registerTerminalResponseWork, registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork, registerTerminalConfirmationWork, diff --git a/server/src/services/chat-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts index 01ccc82250e..29394a24814 100644 --- a/server/src/services/chat-channels/github/maintenance.ts +++ b/server/src/services/chat-channels/github/maintenance.ts @@ -12,7 +12,7 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet options.scheduleMessageProcessing(async () => { try { // Provider I/O must not block Slack, Teams, or ordinary message receipts. - await githubReviewCheckService(db, fetchImpl).processPending(); + await githubReviewCheckService(db, fetchImpl).repairSupersededReviews(); } finally { githubMaintenancePending = false; } diff --git a/server/src/services/chat-channels/github/review-check-state.ts b/server/src/services/chat-channels/github/review-check-state.ts new file mode 100644 index 00000000000..2254a82feb0 --- /dev/null +++ b/server/src/services/chat-channels/github/review-check-state.ts @@ -0,0 +1,133 @@ +import { and, desc, eq, sql } from "drizzle-orm"; +import { chatActions, chatDeliveries, chatEndpoints, chatEndpointResources, + chatGitHubConfigurations, chatGitHubReviews, chatMessageLinks, companies, + heartbeatRuns, type Db } from "@paperclipai/db"; +import type { GitHubReviewEventContext } from "@paperclipai/shared"; + +export async function githubReviewCheckState(db: Db, action: typeof chatActions.$inferSelect) { + const event = action.payload.event as GitHubReviewEventContext; + const [source] = await db + .select({ + endpoint: chatEndpoints, + resource: chatEndpointResources, + configuration: chatGitHubConfigurations.configuration, + companyPrefix: companies.issuePrefix, + }) + .from(chatEndpoints) + .innerJoin(companies, eq(companies.id, chatEndpoints.companyId)) + .innerJoin( + chatGitHubConfigurations, + eq(chatGitHubConfigurations.endpointId, chatEndpoints.id), + ) + .innerJoin( + chatEndpointResources, + eq(chatEndpointResources.endpointId, chatEndpoints.id), + ) + .where( + and( + eq(chatEndpoints.id, action.endpointId), + eq(chatEndpoints.companyId, action.companyId), + eq(chatEndpointResources.providerResourceId, event.repository), + ), + ); + if ( + !source || + !source.configuration.toolsEnabled || + !["active", "verifying"].includes(source.endpoint.status) || + !source.resource.enabled || + source.resource.availability !== "available" || + String(source.resource.metadata?.providerRepositoryId) !== + event.repositoryId || + ((source.endpoint.setup as { runtimeGeneration?: number }) + .runtimeGeneration ?? 0) !== action.payload.runtimeGeneration + ) { + return { disposition: "connection_changed" as const }; + } + const [review] = await db + .select() + .from(chatGitHubReviews) + .where( + and( + eq(chatGitHubReviews.endpointId, action.endpointId), + eq(chatGitHubReviews.repositoryId, event.repositoryId), + eq(chatGitHubReviews.pullNumber, event.pullNumber), + eq(chatGitHubReviews.headSha, event.headSha), + ), + ) + .orderBy(desc(chatGitHubReviews.createdAt)) + .limit(1); + if (review?.assessment) return { disposition: "assessment_owns_check" as const, reviewId: review.id }; + let run = review?.runId + ? ( + await db + .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, review.runId)) + )[0] + : undefined; + if (review && !run) { + const [sourceRun] = await db + .select({ run: { id: heartbeatRuns.id, status: heartbeatRuns.status } }) + .from(chatDeliveries) + .innerJoin( + chatMessageLinks, + eq(chatMessageLinks.deliveryId, chatDeliveries.id), + ) + .innerJoin( + heartbeatRuns, + and( + eq(heartbeatRuns.companyId, action.companyId), + eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), + sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${chatMessageLinks.commentId}::text`, + ), + ) + .where( + and( + eq(chatDeliveries.endpointId, action.endpointId), + sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`, + ), + ) + .orderBy(desc(heartbeatRuns.createdAt)) + .limit(1); + run = sourceRun?.run; + } + const [delivery] = await db + .select({ state: chatDeliveries.state }) + .from(chatDeliveries) + .where( + and( + eq(chatDeliveries.endpointId, action.endpointId), + sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`, + ), + ) + .limit(1); + const admissionEnded = + !review && + (delivery?.state === "filtered" || delivery?.state === "failed"); + const terminal = + !!run && + [ + "succeeded", + "failed", + "cancelled", + "timed_out", + "interrupted", + ].includes(run.status); + const state: (typeof chatGitHubReviews.$inferSelect)["state"] = + (!action.payload.admitted && !review) || admissionEnded + ? "manual_required" + : terminal + ? run!.status === "succeeded" + ? "incomplete" + : "error" + : run?.status === "running" + ? "running" + : "queued"; + const status = + state === "running" + ? "in_progress" + : state === "queued" + ? "queued" + : "completed"; + return { disposition: "publish" as const, source, review, run, terminal, state, status }; +} diff --git a/server/src/services/chat-channels/github/review-check-work.ts b/server/src/services/chat-channels/github/review-check-work.ts new file mode 100644 index 00000000000..d3e951d0aa3 --- /dev/null +++ b/server/src/services/chat-channels/github/review-check-work.ts @@ -0,0 +1,79 @@ +import { and, asc, eq, gt, or, sql } from "drizzle-orm"; +import { chatActions, chatEndpointLeases, type Db } from "@paperclipai/db"; +import type { GitHubReviewEventContext } from "@paperclipai/shared"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; +import { githubReviewCheckState } from "./review-check-state.js"; + +type Scope = { companyId: string; endpointId?: string; repositoryId?: string; number?: number }; +type Entry = Scope & { id: string; dueAt: number }; + +export async function notifyGitHubReviewChecks(tx: object, scope: Scope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify(scope)); +} + +export function registerGitHubReviewCheckWork( + db: Db, + coordinator: ReturnType, + dispatch: (id: string, canContinue: () => boolean) => Promise, + canRun: () => boolean, +) { + async function page(scope?: Scope, cursor?: string, actionId?: string) { + return db.select({ action: chatActions, + 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:' || (${chatActions.payload}->'event'->>'repositoryId') + || ':' || (${chatActions.payload}->'event'->>'pullNumber'))`, + }).from(chatActions).where(and(eq(chatActions.kind, "github_review_check"), + or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)), + scope ? eq(chatActions.companyId, scope.companyId) : undefined, + scope?.endpointId ? eq(chatActions.endpointId, scope.endpointId) : undefined, + scope?.repositoryId ? sql`${chatActions.payload}->'event'->>'repositoryId' = ${scope.repositoryId}` : undefined, + scope?.number ? sql`${chatActions.payload}->'event'->>'pullNumber' = ${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 { action } = row; + const event = action.payload.event as GitHubReviewEventContext; + let dueAt = action.result?.retryAt ? Date.parse(String(action.result.retryAt)) : 0; + if (!Number.isFinite(dueAt)) throw new Error("GitHub review check has an invalid retry deadline"); + if (action.status === "received") { + const current = await githubReviewCheckState(db, action); + if (current.disposition !== "publish" || current.state !== action.result?.state) dueAt = 0; + } + if (row.leaseExpiresAt) dueAt = Math.max(dueAt, Date.parse(row.leaseExpiresAt)); + return { id: action.id, companyId: action.companyId, endpointId: action.endpointId, + repositoryId: event.repositoryId, number: event.pullNumber, dueAt }; + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubReviewChecks, + decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (value: Entry, scope?: Scope) => !scope || (value.companyId === scope.companyId && + (!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 review check recovery stopped"); + entries.push(await entry(row)); + } + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].action.id; + } + throw signal?.reason ?? new Error("GitHub review check 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 dispatch(current.id, () => canRun() && !signal?.aborted); + const [remaining] = await page(undefined, undefined, value.id); + return remaining ? entry(remaining) : null; + }, + }, canRun); +} diff --git a/server/src/services/chat-channels/github/task-preparation.ts b/server/src/services/chat-channels/github/task-preparation.ts index d44ce61de02..6d5988d1990 100644 --- a/server/src/services/chat-channels/github/task-preparation.ts +++ b/server/src/services/chat-channels/github/task-preparation.ts @@ -1,3 +1,4 @@ +import { notifyGitHubReviewChecks } from "./review-check-work.js"; import { agents, chatDeliveries, chatGitHubReviews } from "@paperclipai/db"; import { and, eq, sql } from "drizzle-orm"; import { LOW_TRUST_REVIEW_PRESET, type GitHubReviewEventContext, type GitHubIssueEventContext, type GitHubReviewPolicy } from "@paperclipai/shared"; @@ -41,6 +42,8 @@ export async function retainGitHubDeliveryAuthority( } export async function retainGitHubReviewReceipt(taskTx: DbOrTransaction, endpoint: EndpointRow, issueId: string, deliveryId: string, githubAutomatic: AutomaticRequest): Promise { + await notifyGitHubReviewChecks(taskTx, { companyId: endpoint.companyId, endpointId: endpoint.id, + repositoryId: githubAutomatic.context.repositoryId, number: githubAutomatic.context.pullNumber }); await taskTx.insert(chatGitHubReviews).values({ companyId: endpoint.companyId, endpointId: endpoint.id, issueId: issueId, repositoryId: githubAutomatic.context.repositoryId, repository: githubAutomatic.context.repository, diff --git a/server/src/services/chat-github-checks.ts b/server/src/services/chat-github-checks.ts index a47cf3b1021..c9fe43534c7 100644 --- a/server/src/services/chat-github-checks.ts +++ b/server/src/services/chat-github-checks.ts @@ -1,18 +1,17 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { notifyGitHubReviewChecks, registerGitHubReviewCheckWork } from "./chat-channels/github/review-check-work.js"; +import { githubReviewCheckState } from "./chat-channels/github/review-check-state.js"; import { GitHubPublicationLeaseLost, withGitHubPublicationLease, } from "./chat-github-publication-lease.js"; -import { and, desc, eq, inArray, isNull, or, sql } from "drizzle-orm"; +import { and, eq, inArray, isNull, or, sql } from "drizzle-orm"; import { chatActions, chatDeliveries, chatEndpoints, - chatEndpointResources, chatGitHubConfigurations, chatGitHubReviews, - chatMessageLinks, - companies, - heartbeatRuns, type Db, } from "@paperclipai/db"; import type { GitHubReviewEventContext } from "@paperclipai/shared"; @@ -41,27 +40,32 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { ...config.configuration.repositories[event.repositoryId], }; if (policy.ratingThreshold === null && !admitted) return; - await db - .insert(chatActions) - .values({ - companyId: endpoint.companyId, - endpointId: endpoint.id, - kind: "github_review_check", - providerActionId: `github-check:${event.deliveryId}`, - payload: { - version: 1, - event, - admitted, - reason, - runtimeGeneration: - (endpoint.setup as { runtimeGeneration?: number }) - .runtimeGeneration ?? 0, - }, - status: "received", - }) - .onConflictDoNothing(); + await db.transaction(async tx => { + await notifyGitHubReviewChecks(tx, { companyId: endpoint.companyId, endpointId: endpoint.id, + repositoryId: event.repositoryId, number: event.pullNumber }); + await tx + .insert(chatActions) + .values({ + companyId: endpoint.companyId, + endpointId: endpoint.id, + kind: "github_review_check", + providerActionId: `github-check:${event.deliveryId}`, + payload: { + version: 1, + event, + admitted, + reason, + runtimeGeneration: + (endpoint.setup as { runtimeGeneration?: number }) + .runtimeGeneration ?? 0, + }, + status: "received", + }) + .onConflictDoNothing(); + }); } - async function dispatch(id: string) { + async function dispatch(id: string, canContinue = () => true) { + if (!canContinue()) return; const [action] = await db .select() .from(chatActions) @@ -74,6 +78,11 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { .limit(1); if (!action || ["processed", "cancelled"].includes(action.status)) return; const event = action.payload.event as GitHubReviewEventContext; + const guardedFetch: typeof fetch = (input, init) => { + if (!canContinue()) throw new GitHubPublicationLeaseLost(); + return fetchImpl(input, init); + }; + if (!canContinue()) return; await withGitHubPublicationLease( db, { @@ -82,7 +91,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { repositoryId: event.repositoryId, number: event.pullNumber, }, - fetchImpl, + guardedFetch, async (lease) => { const [fresh] = await db .select() @@ -91,155 +100,19 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { if (!fresh || ["processed", "cancelled"].includes(fresh.status)) return; let attempts = Number(fresh.result?.attempts ?? 0); try { - const [source] = await db - .select({ - endpoint: chatEndpoints, - resource: chatEndpointResources, - configuration: chatGitHubConfigurations.configuration, - companyPrefix: companies.issuePrefix, - }) - .from(chatEndpoints) - .innerJoin(companies, eq(companies.id, chatEndpoints.companyId)) - .innerJoin( - chatGitHubConfigurations, - eq(chatGitHubConfigurations.endpointId, chatEndpoints.id), - ) - .innerJoin( - chatEndpointResources, - eq(chatEndpointResources.endpointId, chatEndpoints.id), - ) - .where( - and( - eq(chatEndpoints.id, action.endpointId), - eq(chatEndpoints.companyId, action.companyId), - eq(chatEndpointResources.providerResourceId, event.repository), - ), - ); - if ( - !source || - !source.configuration.toolsEnabled || - !["active", "verifying"].includes(source.endpoint.status) || - !source.resource.enabled || - source.resource.availability !== "available" || - String(source.resource.metadata?.providerRepositoryId) !== - event.repositoryId || - ((source.endpoint.setup as { runtimeGeneration?: number }) - .runtimeGeneration ?? 0) !== action.payload.runtimeGeneration - ) { - await lease.commit(async (tx) => { - await tx - .update(chatActions) - .set({ - status: "cancelled", - result: { code: "connection_changed" }, - updatedAt: new Date(), - }) - .where(eq(chatActions.id, id)); + const currentState = await githubReviewCheckState(db, fresh); + if (currentState.disposition !== "publish") { + await lease.commit(async tx => { + await tx.update(chatActions).set({ + status: currentState.disposition === "connection_changed" ? "cancelled" : "processed", + result: { code: currentState.disposition, + ...(currentState.disposition === "assessment_owns_check" ? { reviewId: currentState.reviewId } : {}) }, + updatedAt: new Date(), + }).where(eq(chatActions.id, id)); }); return; } - const [review] = await db - .select() - .from(chatGitHubReviews) - .where( - and( - eq(chatGitHubReviews.endpointId, action.endpointId), - eq(chatGitHubReviews.repositoryId, event.repositoryId), - eq(chatGitHubReviews.pullNumber, event.pullNumber), - eq(chatGitHubReviews.headSha, event.headSha), - ), - ) - .orderBy(desc(chatGitHubReviews.createdAt)) - .limit(1); - if (review?.assessment) { - // The assessment outbox owns the terminal score and its retries. - await lease.commit(async (tx) => { - await tx - .update(chatActions) - .set({ - status: "processed", - result: { - code: "assessment_owns_check", - reviewId: review.id, - }, - updatedAt: new Date(), - }) - .where(eq(chatActions.id, id)); - }); - return; - } - let run = review?.runId - ? ( - await db - .select({ id: heartbeatRuns.id, status: heartbeatRuns.status }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, review.runId)) - )[0] - : undefined; - if (review && !run) { - const [sourceRun] = await db - .select({ run: { id: heartbeatRuns.id, status: heartbeatRuns.status } }) - .from(chatDeliveries) - .innerJoin( - chatMessageLinks, - eq(chatMessageLinks.deliveryId, chatDeliveries.id), - ) - .innerJoin( - heartbeatRuns, - and( - eq(heartbeatRuns.companyId, action.companyId), - eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId), - sql`${heartbeatRuns.contextSnapshot}->>'wakeCommentId' = ${chatMessageLinks.commentId}::text`, - ), - ) - .where( - and( - eq(chatDeliveries.endpointId, action.endpointId), - sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`, - ), - ) - .orderBy(desc(heartbeatRuns.createdAt)) - .limit(1); - run = sourceRun?.run; - } - const [delivery] = await db - .select({ state: chatDeliveries.state }) - .from(chatDeliveries) - .where( - and( - eq(chatDeliveries.endpointId, action.endpointId), - sql`${chatDeliveries.normalizedEvent}->'githubAutomatic'->'context'->>'deliveryId' = ${event.deliveryId}`, - ), - ) - .limit(1); - const admissionEnded = - !review && - (delivery?.state === "filtered" || delivery?.state === "failed"); - const terminal = - !!run && - [ - "succeeded", - "failed", - "cancelled", - "timed_out", - "interrupted", - ].includes(run.status); - const state = - (!action.payload.admitted && !review) || admissionEnded - ? "manual_required" - : terminal - ? run!.status === "succeeded" - ? "incomplete" - : "error" - : run?.status === "running" - ? "running" - : "queued"; - const status = - state === "running" - ? "in_progress" - : state === "queued" - ? "queued" - : "completed"; + const { source, review, run, terminal, state, status } = currentState; const policy = review?.policySnapshot ?? { ...source.configuration.defaults, ...source.configuration.repositories[event.repositoryId], @@ -389,7 +262,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { .where(eq(chatActions.id, id)); }); } catch (error) { - if (error instanceof GitHubPublicationLeaseLost) return; + if (error instanceof GitHubPublicationLeaseLost || !canContinue()) return; attempts++; await lease.commit(async (tx) => { await tx @@ -412,7 +285,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { }, ); } - async function processPending(limit = 10) { + async function repairSupersededReviews() { // Check delivery and task admission run independently. A rapid push can // cancel an obsolete check before admission inserts its review projection. // Reconcile that late row from the durable cancellation receipt, without @@ -447,6 +320,9 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { ) is not null`, ), ); + } + async function processPending(limit = 10) { + await repairSupersededReviews(); const rows = await db .select({ id: chatActions.id }) .from(chatActions) @@ -470,5 +346,8 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { .limit(limit); for (const row of rows) await dispatch(row.id); } - return { enqueue, processPending }; + return { enqueue, processPending, repairSupersededReviews, + registerCheckWork: (coordinator: ReturnType, canRun: () => boolean) => + registerGitHubReviewCheckWork(db, coordinator, dispatch, canRun), + }; } diff --git a/server/src/services/chat-github-publication-lease.ts b/server/src/services/chat-github-publication-lease.ts index 994bb8e9174..a6e5976a7cc 100644 --- a/server/src/services/chat-github-publication-lease.ts +++ b/server/src/services/chat-github-publication-lease.ts @@ -115,6 +115,7 @@ export async function withGitHubPublicationLease( clearInterval(timer); await renewal; await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewChecks, JSON.stringify(scope)); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify(scope)); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify(scope)); diff --git a/server/src/services/chat-github-reviews.ts b/server/src/services/chat-github-reviews.ts index 06102e8df4b..cdddc6b3eae 100644 --- a/server/src/services/chat-github-reviews.ts +++ b/server/src/services/chat-github-reviews.ts @@ -1,3 +1,4 @@ +import { notifyGitHubReviewChecks } from "./chat-channels/github/review-check-work.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { githubRunReplyState } from "./chat-run-publications.js"; import { stageReceiptReactionRemovals } from "./chat-receipt-reactions.js"; @@ -350,16 +351,20 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { .limit(1); if (prior && prior.runId === source.run.id) return prior; if (prior && !prior.runId && prior.deliveryId === source.delivery.id) { - const [claimed] = await db - .update(chatGitHubReviews) - .set({ runId: source.run.id, state: "running", updatedAt: new Date() }) - .where( - and( - eq(chatGitHubReviews.id, prior.id), - isNull(chatGitHubReviews.runId), - ), - ) - .returning(); + const [claimed] = await db.transaction(async tx => { + await notifyGitHubReviewChecks(tx, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, + repositoryId: source.repositoryId, number: source.number }); + return tx + .update(chatGitHubReviews) + .set({ runId: source.run.id, state: "running", updatedAt: new Date() }) + .where( + and( + eq(chatGitHubReviews.id, prior.id), + isNull(chatGitHubReviews.runId), + ), + ) + .returning(); + }); if (claimed) return claimed; return reviewForHead(source, pull); } @@ -407,26 +412,30 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { labels: pull.labels.map((l) => l.name), ...(automatic ? automatic.context : {}), }; - const [review] = await db - .insert(chatGitHubReviews) - .values({ - companyId: source.endpoint.companyId, - endpointId: source.endpoint.id, - issueId: source.issue.id, - runId: source.run.id, - repositoryId: source.repositoryId, - repository: source.repository, - pullNumber: source.number, - headSha: pull.head.sha, - deliveryId: `run:${source.run.id}:${pull.head.sha}`, - configurationRevision: - manual?.revision ?? automatic?.revision ?? source.config.revision, - policySnapshot: manual?.policy ?? automatic?.policy ?? source.policy, - event: context, - state: "running", - }) - .onConflictDoNothing() - .returning(); + const [review] = await db.transaction(async tx => { + await notifyGitHubReviewChecks(tx, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, + repositoryId: source.repositoryId, number: source.number }); + return tx + .insert(chatGitHubReviews) + .values({ + companyId: source.endpoint.companyId, + endpointId: source.endpoint.id, + issueId: source.issue.id, + runId: source.run.id, + repositoryId: source.repositoryId, + repository: source.repository, + pullNumber: source.number, + headSha: pull.head.sha, + deliveryId: `run:${source.run.id}:${pull.head.sha}`, + configurationRevision: + manual?.revision ?? automatic?.revision ?? source.config.revision, + policySnapshot: manual?.policy ?? automatic?.policy ?? source.policy, + event: context, + state: "running", + }) + .onConflictDoNothing() + .returning(); + }); if (review) { await githubReviewCheckService(db, fetchImpl).enqueue( source.endpoint, diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index b277485b7e2..8ab9636b6af 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.chatGitHubReviewChecks, JSON.stringify({ companyId: scope.companyId })); 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 diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 20bb0a4b808..afe718a27ec 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", + chatGitHubReviewChecks: "chat-github-review-checks", chatGitHubTerminalResponses: "chat-github-terminal-responses", chatGitHubAcknowledgements: "chat-github-acknowledgements", chatFailedRunRetries: "chat-failed-run-retries",