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 992cca11f32..d958711d44e 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2438,3 +2438,20 @@ 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. + +Slice 37, codex/sleep-37-github-review-projection, separates late stale-head +review projection repair from periodic chat maintenance. Existing cancellation +receipts and exact admitted deliveries authorize the repair. Committed receipt, +cancellation, and relevant review-state writes wake the provider-owned worker; +assessment ownership is rechecked atomically. The obsolete provider maintenance +wrapper is removed. Explicit manual recovery and the independent missed-webhook +lane remain. No table or Cloud wake contract is added. Ten new real-database +cases plus legacy/startup coverage pass; final scalar-query acceptance and remote +gates remain pending. See the owner audit for fixture failures, repairs, index +plan evidence, and lifecycle limits. Other original inventory items and earlier +stack gates remain open. + +Slice 36 final remote gates: all 53 checks passed on 01efd3f4f7, with fresh +Greptile 5/5 and no review threads. Slice 37 final scalar-query selection passed +20 tests; workspace typecheck and build passed. Full local acceptance and remote +PR gates are still pending. No PR in this stack was merged. diff --git a/doc/plans/2026-10-25-github-review-projection-sleep-audit.md b/doc/plans/2026-10-25-github-review-projection-sleep-audit.md new file mode 100644 index 00000000000..6c119f99f70 --- /dev/null +++ b/doc/plans/2026-10-25-github-review-projection-sleep-audit.md @@ -0,0 +1,191 @@ +# GitHub stale-head review projection repair + +Date: 2026-10-25. Parent: GitHub review-check migration #17662. + +## Obligation and owner + +The GitHub check dispatcher can record `cancelled/stale_head` before task +admission creates its review row. A review in `queued` or `running` with no +assessment must then become `superseded`, but only when the cancellation belongs +to that exact admitted delivery. Same PR/head alone is not enough authority. + +Existing reviews, check actions, and deliveries are durable truth. This is local +projection repair, not provider reconciliation. It needs no provider HTTP, +periodic deadline, or new queue table. The GitHub provider directory owns the +predicate and repair; the shared coordinator owns coalescing, fairness, retries, +drain, startup reconstruction, and shutdown. + +## Producer audit + +- `githubReviewCheckService.dispatch`: stale-head cancellation commits the action. + Signal before the action write, even if no review existed during dispatch. +- `retainGitHubReviewReceipt`: task admission inserts the automatic review in the + caller-owned transaction. Signal before insertion, preserving outer settlement. +- `githubChatReviewService.reviewForHead`: binding an unbound automatic review can + set it back to running. Signal before that update. Newly inserted manual review + identities use `run:...` and cannot match an admitted delivery UUID. +- Check publication can set a review to running. Signal before that review update + to cover an existing cancellation for its exact delivery. +- Assessment writes remove eligibility. Publication receipts/check metadata do + not create it. Supersession removes it. Delivery authority updates preserve the + original automatic context; task admission creates the review after the context. + +## Correctness and lifecycle + +Discover only matching unfinished repairs, with bounded keyset pages and shared +fair dispatch. Recheck the complete pending predicate in the UPDATE so a competing +assessment or state transition cannot be overwritten from a stale selection. +A failed selection or update retains a retry through the existing coordinator. +No pending repair means no recurring SQL. Configured endpoints, manual reviews, +assessed reviews, and settled cancellation history are passive. + +Signals are process-local to a database owner and its child clients. Independent +SQL writers or another process require an explicit integration; startup rebuilds +already committed work. This does not extend Cloud wake coverage, relax idle +safety, or prove host/database suspension. + +## Scope + +Remove this duty from the periodic GitHub maintenance wrapper. Retire the empty +wrapper and its publication-maintenance wiring while preserving the independent +GitHub missed-webhook recovery lane. Preserve explicit/manual combined check +recovery. Other server maintenance remains in the original stack inventory. + +## Evidence + +Ten new real-PostgreSQL cases cover real webhook admission after cancellation, +cancellation after a held remote-head read and intervening admission, nested +receipt commit/rollback, restart/drain/standby, failed-update retry across drain, +assessment ownership under an observed PostgreSQL row-lock wait (with and +without concurrent shutdown), and 106 pending repairs plus 105 manual rows. +The existing late-projection regression still exercises the explicit recovery API. + +The first selection passed eight and failed two cases. The nested-commit +assertion had the default one-second timeout, matching rather than exceeding +the shared worker's one-second coalescing interval. It now uses the established +five-second observation allowance. Warm standby correctly retains an unresolved +intent hold until reconciliation is admitted; the fixture now asserts that hold +while also proving no passive SQL. No coordinator or production delay changed. +After repair, all 20 worker/legacy/startup cases passed in 24.72 seconds and +again in 30.61 seconds after query/index changes. + +Workspace typecheck and build passed before the final scalar-query refinement; +fresh acceptance for that final refinement is pending. Migration generation, +safety checks, application, replay, and index validity passed. The generated +migration adds two concurrent, idempotent indexes and retains five snapshots. + +A rollback-only PostgreSQL plan seeded 10,000 each of completed reviews, manual +queued reviews, processed checks, unrelated cancelled checks, and deliveries. +The original joined EXISTS query used the receipt index only for company and +endpoint. An unrelated prefix forced 10,001 cancellation reads, 32,382 shared +buffers and about 34 ms. A nested EXISTS form was flattened into a hash scan of +retained cancellations. The final scalar receipt lookup uses all three index +keys, the delivery primary key, and the partial pending-review index: one result, +21 shared buffers, about 0.062 ms in the recorded fixture. This is a query-plan +check, not a whole-worker or fleet benchmark. + +The canonical lower-case UUID guard preserves the old delivery.id::text equality: +manual run identities and noncanonical strings never matched that predicate. +CASE guards the cast even if PostgreSQL reorders predicates. The receipt lookup +is limited to one row and runs with the review's company and endpoint authority. +The candidate partial index excludes assessed, terminal, and manual history. + +Final scalar-query verification: all 20 selected worker, legacy regression, and +startup tests passed in 33.62 seconds. Workspace typecheck and build passed on +the same source. Full chat/publication/startup acceptance is running in a fresh +database; its result and remote PR checks remain required. Parent #17662 at +01efd3f4f7 has all 53 checks passing, fresh Greptile 5/5, and no review threads. + + +## Current-parent acceptance (October 28) + +The original query, authority checks, update predicate, and producer/lifecycle +changes remain intact. The existing branch was merged normally with published +review-check parent `2bba08403b195bf646eda48992b46d6f5cebf756`. Complete domain +and schema edits and all unrelated parent blobs were preserved. The historical +snapshot rename/delete conflict was resolved to the exact parent snapshot bytes. +Only this branch's unlanded `0397_fearless_supernaut` migration artifacts were +removed before canonical regeneration; all parent SQL and journal entries remain. + +Migration `0403_natural_cerise.sql` creates the same two partial indexes with +concurrent, idempotent creation. Both definitions match the original migration. +All retained parent snapshot bytes are unchanged. Canonical generation retains +the five snapshots 0399 through 0403. + +At `dd6bca0485c3444d6e3d0ee10ec4aa651e801392`, all 274 complete GitHub workflow +cases and all 13 startup cases passed: 287 distinct cases. The GitHub command +filtered 1,485 unrelated cases. It used a new, pre-migrated PostgreSQL database. +Coverage includes cancellation before/after real admission, outer commit and +rollback, restart, drain and standby, update failure, concurrent assessment under +an observed database row lock, shutdown during that wait, multi-page repair, +manual-history exclusion, and settled-history quiet. Database build, server +production typecheck/build, module boundaries, and target migration order passed. + +A populated parent database upgraded from 0402 to 0403 with all 401 previous +migration receipts unchanged and exactly one new receipt. Replay added none. +Both indexes were ready and valid on fresh and upgraded PostgreSQL databases. +The earlier query-plan measurements above retain their historical revision scope; +no new fleet or whole-worker performance measurement is claimed. + +The audit-only follow-up preserves the accepted source, tests, and migration +hashes. Fresh remote checks and exact-current-head Greptile review remain required. +This acceptance does not prove deployed host/database suspension or extend Cloud +alarm coverage. The separate missed-webhook obligation remains for the next slice. + + +## Browser CI clipboard read repair (October 28) + +Head `1453a4447c0d4d33f713ae3e00734c7950f95426` received fresh Greptile 5/5 with +no findings. Its browser shard 7 failed twice in the lost-creation-response YAML +journey: the clipboard helper copied page text beginning with “Skip to Main +Content” and tried to parse it as YAML. The second failure followed the single +allowed unchanged-code retry. All three initial isolated local repetitions passed +with retries disabled and original budgets; that did not establish resolution. +The entire UI, failing test, Playwright config, and lockfile match the green parent. + +The CI failure artifact shows the main region focused and the YAML editor not +focused. A one-shot native focus change to main during Copy, with whole-page +selection, reproduced the identical parser failure in the real browser. The exact +actor and timing that changed focus in CI were not captured; no infrastructure +cause is inferred from the earlier local passes. + +The shared test read helper now clears stale clipboard data, clicks the actual +editor, and repeats the read-only copy gesture within the existing expectation +budget until a fresh canonical YAML header is present. YAML parsing still runs, +and every journey body, persistence assertion, command counter, and test deadline +is unchanged. This changes the browser fixture, not production background work. + +The controlled focus-loss probe failed before the repair and passed afterward. +The temporary fault was removed before acceptance. All six complete agent YAML +configuration journeys then passed with zero retries in 50.9 seconds. This includes +real resource writes, partial/lost responses, non-replayed creation commands, +conflict handling, destination instructions, and revision restore. + +Only the central test read helper and this verification note change after the +reviewed head. Entire server, package, and UI source trees remain identical. +The earlier 287-case backend acceptance therefore retains the same source bytes. +Fresh current-head CI and Greptile review are required for this repair; the old +head's score does not satisfy that gate. Both earlier CI failures remain recorded +in run 40777754306, including attempts 1 and 2, and its failed-job artifacts. + + +## Ponytail startup boundary correction + +All CI and Greptile checks passed on `b3c81baebf1787ffb200bdd55533b422c2687033`, +but Ponytail's completed 4/5 review said “Ship after fixing 1.” That conditional +verdict blocked the merge. Its required finding was valid: one source-wiring test +used the removed `unsubscribeChatPublicationSignals` marker, so its end index was +-1 and its slice inspected almost the whole application file. + +Explicit boundary assertions reproduced that defect: the selected test failed +because -1 was not after its start. The test now stops at the existing +`let chatPublicationTimer:` declaration and requires a nonnegative start and an +end after that start. All 13 startup/hostname cases pass with this correction. +No production behavior, deadline, or browser assertion changes in this follow-up. +The YAML helper and its six-case acceptance remain unchanged. + +The review's second item was optional cleanup of the no-op coordinator lane. +Existing dependent #17666 removes the whole coordinator and the last chat interval; +this PR keeps its narrower scope. Fresh review and CI are required for the corrected +head. A neutral review check or a confidence score does not override a conditional +ship verdict or an unresolved required finding. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index f9a2c6032a0..2b0f0663078 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -554,6 +554,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0400_absent_phantom_reporter.sql", "0401_yummy_ultimates.sql", "0402_ancient_malice.sql", + "0403_natural_cerise.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0403_natural_cerise.sql b/packages/db/src/migrations/0403_natural_cerise.sql new file mode 100644 index 00000000000..799975e1a09 --- /dev/null +++ b/packages/db/src/migrations/0403_natural_cerise.sql @@ -0,0 +1,5 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_stale_head_idx" ON "chat_actions" USING btree ("company_id","endpoint_id",("payload"->'event'->>'deliveryId')) WHERE "chat_actions"."kind" = 'github_review_check' + and "chat_actions"."status" = 'cancelled' and "chat_actions"."result"->>'code' = 'stale_head';--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_github_reviews_projection_work_idx" ON "chat_github_reviews" USING btree ("id") WHERE "chat_github_reviews"."state" in ('queued', 'running') + and "chat_github_reviews"."assessment" is null + and "chat_github_reviews"."delivery_id" ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0398_snapshot.json b/packages/db/src/migrations/meta/0403_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0398_snapshot.json rename to packages/db/src/migrations/meta/0403_snapshot.json index 2d38694b8c3..957177038ec 100644 --- a/packages/db/src/migrations/meta/0398_snapshot.json +++ b/packages/db/src/migrations/meta/0403_snapshot.json @@ -1,6 +1,6 @@ { - "id": "7f056258-4b1a-4828-b950-33f070fc0c47", - "prevId": "c7947622-f079-44c9-a57a-5a11d1bec5e8", + "id": "e21e38de-5a82-473f-b81e-e96eff5ef4c8", + "prevId": "6ce76555-38d9-4074-873f-db02c546e2c6", "version": "7", "dialect": "postgresql", "tables": { @@ -9624,6 +9624,28 @@ } }, "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": [ @@ -9666,6 +9688,82 @@ "concurrently": false, "method": "btree", "with": {} + }, + "chat_actions_github_stale_head_idx": { + "name": "chat_actions_github_stale_head_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "endpoint_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "(\"payload\"->'event'->>'deliveryId')", + "asc": true, + "isExpression": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_actions\".\"kind\" = 'github_review_check'\n and \"chat_actions\".\"status\" = 'cancelled' and \"chat_actions\".\"result\"->>'code' = 'stale_head'", + "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": { @@ -12093,6 +12191,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": [ @@ -13105,6 +13237,22 @@ } }, "indexes": { + "chat_github_reviews_projection_work_idx": { + "name": "chat_github_reviews_projection_work_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"chat_github_reviews\".\"state\" in ('queued', 'running')\n and \"chat_github_reviews\".\"assessment\" is null\n and \"chat_github_reviews\".\"delivery_id\" ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'", + "concurrently": false, + "method": "btree", + "with": {} + }, "chat_github_reviews_delivery_uq": { "name": "chat_github_reviews_delivery_uq", "columns": [ @@ -30538,6 +30686,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 60652c8d108..3610fc5c61c 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2808,6 +2808,13 @@ "when": 1793223470886, "tag": "0402_ancient_malice", "breakpoints": true + }, + { + "idx": 403, + "version": "7", + "when": 1793224779622, + "tag": "0403_natural_cerise", + "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 165bf7f86df..06bf31d12df 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_stale_head_idx").on(table.companyId, table.endpointId, + sql`(${table.payload}->'event'->>'deliveryId')`).where(sql`${table.kind} = 'github_review_check' + and ${table.status} = 'cancelled' and ${table.result}->>'code' = 'stale_head'`), 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' diff --git a/packages/db/src/schema/chat_github.ts b/packages/db/src/schema/chat_github.ts index ebd1928ccaf..65ff9a685d4 100644 --- a/packages/db/src/schema/chat_github.ts +++ b/packages/db/src/schema/chat_github.ts @@ -159,6 +159,9 @@ export const chatGitHubReviews = pgTable( }, (t) => [ unique("chat_github_reviews_company_id_uq").on(t.companyId, t.id), + index("chat_github_reviews_projection_work_idx").on(t.id).where(sql`${t.state} in ('queued', 'running') + and ${t.assessment} is null + and ${t.deliveryId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'`), foreignKey({ columns: [t.companyId, t.endpointId], foreignColumns: [chatEndpoints.companyId, chatEndpoints.id], diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 14812e22085..425e53e98e4 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -74,7 +74,11 @@ describe("createChatReconciliationCoordinator", () => { it("keeps receiver and voice inventory out of periodic chat maintenance", () => { const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8"); - const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals")); + const maintenanceStart = source.indexOf("const chatReconciliation ="); + const maintenanceEnd = source.indexOf("let chatPublicationTimer:", maintenanceStart); + expect(maintenanceStart).toBeGreaterThanOrEqual(0); + expect(maintenanceEnd).toBeGreaterThan(maintenanceStart); + const maintenance = source.slice(maintenanceStart, maintenanceEnd); expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()"); expect(maintenance).not.toContain("chatChannels.voice."); expect(maintenance).not.toContain("enqueueSafeNativeChatProgress"); @@ -95,12 +99,11 @@ describe("createChatReconciliationCoordinator", () => { expect(publicationMaintenance).not.toContain("reconcileTerminalConfirmationActions("); expect(source).toContain("chatChannels.registerInteractionContinuationWork(deliveryWork,"); expect(source).toContain("chatChannels.registerGitHubAcknowledgementWork(deliveryWork,"); - const github = readFileSync(new URL("../services/chat-channels/github/maintenance.ts", import.meta.url), "utf8"); - expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processPending()"); - expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()"); + expect(chat).not.toContain("scheduleGitHubMaintenance"); + expect(source).toContain("registerGitHubReviewProjectionWork(db, deliveryWork,"); + expect(source).not.toContain("chatChannels.processPublicationMaintenance()"); 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,"); }); @@ -137,10 +140,10 @@ describe("createChatReconciliationCoordinator", () => { expect(source).toContain("registerChatDeliveryWork(deliveryWork, chatChannels,"); expect(source).toContain("registerChatActionWork(deliveryWork, chatChannels,"); const maintenance = source.slice( - source.indexOf("const reconcileChatPublicationMaintenance ="), source.indexOf("const chatReconciliation ="), + source.indexOf("let chatPublicationTimer:"), ); - expect(maintenance).toContain("chatChannels.processPublicationMaintenance()"); + expect(maintenance).not.toContain("chatChannels.processPublicationMaintenance()"); expect(maintenance).not.toContain("chatChannels.schedulePendingPublications()"); expect(maintenance).not.toContain("chatChannels.processPendingPublications()"); const workers = readFileSync(new URL("../services/chat-delivery-work.ts", import.meta.url), "utf8"); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 8c062b12a8f..c5b0118fe0d 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,5 @@ +import { registerGitHubReviewProjectionWork } from "../services/chat-channels/github/review-projection-work.js"; +import { retainGitHubReviewReceipt } from "../services/chat-channels/github/task-preparation.js"; 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"; @@ -7483,6 +7485,223 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { } finally { release(); await coordinator.stop(); } }); + async function reviewProjectionFixture(cancelFirst = true) { + const f = await reviewBotFixture(); + const deliveryId = randomUUID(); + const payload = { action: "synchronize", installation: { id: 2468 }, + repository: { id: 97531, full_name: "paperclipai/paperclip", name: "paperclip", owner: { id: 1357, login: "paperclipai" } }, + sender: { id: 42, login: "octocat" }, + pull_request: { number: 82, title: "Rapid pushes", body: "", draft: false, + base: { sha: "a".repeat(40), ref: "master" }, head: { sha: "b".repeat(40) }, + user: { id: 42, login: "octocat", type: "User" }, labels: [] } }; + f.setSupplementalProviderFetch(async input => String(input).endsWith("/pulls/82") + ? Response.json({ ...payload.pull_request, head: { sha: "c".repeat(40) } }) : undefined); + const checks = githubReviewCheckService(db, f.providerFetch); + const [endpoint] = await readChatEndpointsById(f.endpoint.id); + const event = githubAutomaticReviewEvent(payload, deliveryId)!; + await checks.enqueue(endpoint, event, true, "authorized"); + const action = async () => (await db.select().from(chatActions).where(and( + eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "github_review_check"))))[0]; + if (cancelFirst) await checks.processPending(); + const admit = async () => { + expect((await f.service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({ + event: "pull_request", delivery: deliveryId, payload, webhookSecret: f.webhookSecret, + }))).status).toBeLessThan(300); + await expect.poll(async () => (await readChatGitHubReviewsByEndpointId(endpoint.id)).length, + { timeout: 10_000 }).toBe(1); + return (await readChatGitHubReviewsByEndpointId(endpoint.id))[0]; + }; + const read = async (id: string) => (await db.select().from(chatGitHubReviews).where(eq(chatGitHubReviews.id, id)))[0]; + return { ...f, endpoint, event, action, checks, admit, read }; + } + + it("wakes stale GitHub review projection repair from real admission and leaves settled history quiet", async () => { + const f = await reviewProjectionFixture(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + await registerGitHubReviewProjectionWork(db, coordinator, () => true).ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + const review = await f.admit(); + await expect.poll(async () => (await f.read(review.id)).state, { timeout: 5000 }).toBe("superseded"); + await f.service.shutdown(); + const [manual] = await db.insert(chatGitHubReviews).values({ ...review, id: randomUUID(), + deliveryId: "run:" + randomUUID() + ":" + review.headSha, state: "queued" }).returning(); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); } + expect((await f.read(manual.id)).state).toBe("queued"); expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("wakes stale GitHub review projection repair when cancellation commits after the review is admitted", async () => { + const f = await reviewProjectionFixture(false); + let entered!: () => void, release!: () => void; + const enteredHead = new Promise(resolve => { entered = resolve; }); + const hold = new Promise(resolve => { release = resolve; }); + const provider: typeof fetch = async (input, init) => { + if (String(input).endsWith("/pulls/82")) { entered(); await hold; } + return f.providerFetch(input, init); + }; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + let dispatch: Promise | undefined; + try { + await registerGitHubReviewProjectionWork(db, coordinator, () => true).ready; + dispatch = githubReviewCheckService(db, provider).processPending(); + await enteredHead; + const review = await f.admit(); + expect((await f.read(review.id)).state).toBe("queued"); + release(); await dispatch; + await expect.poll(async () => (await f.read(review.id)).state, { timeout: 5000 }).toBe("superseded"); + expect((await f.action()).result).toMatchObject({ code: "stale_head" }); expect(errors).toEqual([]); + } finally { release(); await dispatch; await coordinator.stop(); } + }); + + it("holds stale GitHub review projection receipt repair until the outer commit and discards rollback", async () => { + const f = await reviewProjectionFixture(); + const review = await f.admit(); await f.service.shutdown(); + await db.delete(chatGitHubReviews).where(eq(chatGitHubReviews.id, review.id)); + const receipt = (tx: Parameters[0]>[0]) => retainGitHubReviewReceipt(tx, f.endpoint, + review.issueId, review.deliveryId, { context: review.event, revision: review.configurationRevision, policy: review.policySnapshot }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + let release!: () => void; + let pending: Promise | undefined; + try { + await registerGitHubReviewProjectionWork(db, coordinator, () => true).ready; + await expect(db.transaction(async tx => { await tx.transaction(receipt); throw new Error("rollback projection receipt"); })) + .rejects.toThrow("rollback projection receipt"); + expect(await readChatGitHubReviewsByEndpointId(f.endpoint.id)).toEqual([]); + let entered!: () => void; + const held = new Promise(resolve => { entered = resolve; }); + const wait = new Promise(resolve => { release = resolve; }); + pending = db.transaction(async tx => { await tx.transaction(receipt); entered(); await wait; }); + await held; + expect(await readChatGitHubReviewsByEndpointId(f.endpoint.id)).toEqual([]); + release(); await pending; + await expect.poll(async () => (await readChatGitHubReviewsByEndpointId(f.endpoint.id))[0]?.state, { timeout: 5000 }).toBe("superseded"); + } finally { release?.(); await pending; await coordinator.stop(); } + }); + + it.each(["restart", "drain", "standby"] as const)("recovers stale GitHub review projection through %s", async mode => { + const f = await reviewProjectionFixture(); + let canRun = mode === "restart"; + const errors: unknown[] = []; + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await registerGitHubReviewProjectionWork(db, coordinator, () => canRun).ready; + if (mode === "restart") await coordinator.stop(); + const review = await f.admit(); await f.service.shutdown(); + await vi.advanceTimersByTimeAsync(60_000); + await vi.waitFor(() => { + if (mode === "standby") expect(idleWorkSnapshot().active).toBeGreaterThan(0); + else expect(idleWorkSnapshot().active).toBe(0); + }); + expect((await f.read(review.id)).state).toBe("queued"); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"); + try { await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + canRun = true; + if (mode === "restart") { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerGitHubReviewProjectionWork(db, coordinator, () => true).ready; + } + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect((await f.read(review.id)).state).toBe("superseded"); }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("retains failed stale GitHub review projection repair across drain without passive SQL", async () => { + const f = await reviewProjectionFixture(); const review = await f.admit(); await f.service.shutdown(); + let canRun = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true, onError: error => errors.push(error) }); + const failure = new Error("projection write unavailable"); + const update = vi.spyOn(db, "update").mockImplementationOnce(() => { throw failure; }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await registerGitHubReviewProjectionWork(db, coordinator, () => canRun).ready; + expect(errors).toHaveLength(1); expect((await f.read(review.id)).state).toBe("queued"); + canRun = false; update.mockRestore(); + await vi.advanceTimersByTimeAsync(30_000); await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"); + try { await vi.advanceTimersByTimeAsync(60_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); } + finally { reads.mockRestore(); writes.mockRestore(); } + canRun = true; await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect((await f.read(review.id)).state).toBe("superseded"); }); + expect(errors).toHaveLength(1); + } finally { update.mockRestore(); await coordinator.stop(); vi.useRealTimers(); } + }); + + it.each([false, true])("rechecks assessment ownership after stale GitHub review projection selection waits for a row lock (shutdown: %s)", async shutdown => { + const f = await reviewProjectionFixture(); const review = await f.admit(); await f.service.shutdown(); + const assessment: NonNullable = { reviewedCommit: review.headSha, score: 4, complete: true, + summary: "Reviewed", rationale: "Covered changes", coverage: { reviewedPaths: ["src/a.ts"], omittedPaths: [], limitations: [] }, findings: [] }; + let locked!: () => void, release!: () => void; + const ready = new Promise(resolve => { locked = resolve; }), held = new Promise(resolve => { release = resolve; }); + const pending = db.transaction(async tx => { + await tx.select().from(chatGitHubReviews).where(eq(chatGitHubReviews.id, review.id)).for("update"); + locked(); await held; + await tx.update(chatGitHubReviews).set({ assessment }).where(eq(chatGitHubReviews.id, review.id)); + }); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } }); + const updates = vi.spyOn(db, "update"); + try { + await ready; + const starting = registerGitHubReviewProjectionWork(db, coordinator, () => true).ready; + await expect.poll(() => updates.mock.calls.length, { timeout: 5000 }).toBe(1); + await expect.poll(async () => { + const rows = await db.execute<{ waiting: boolean }>(sql`select exists ( + select 1 from pg_stat_activity where datname = current_database() + and wait_event_type = 'Lock' and query like 'update %chat_github_reviews%') as waiting`); + return rows[0].waiting; + }, { timeout: 5000 }).toBe(true); + let stopped = false; + const stopping = shutdown ? coordinator.stop().then(() => { stopped = true; }) : undefined; + await Promise.resolve(); expect(stopped).toBe(false); + release(); await pending; await starting; await stopping; + expect(await f.read(review.id)).toMatchObject({ state: "queued", assessment }); + expect(coordinator.nextWakeAt()).toBeNull(); + } finally { release(); await pending; await coordinator.stop(); updates.mockRestore(); } + }); + + it("drains more than one page of stale GitHub review projections without touching manual history", async () => { + const f = await reviewProjectionFixture(); const review = await f.admit(); await f.service.shutdown(); + const [delivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, review.deliveryId)); + const action = await f.action(); + const extra = Array.from({ length: 105 }, () => ({ id: randomUUID(), receipt: randomUUID() })); + await db.insert(chatDeliveries).values(extra.map(item => ({ ...delivery, id: item.id, providerEventId: item.receipt, + deduplicationKey: item.receipt, normalizedEvent: { ...delivery.normalizedEvent, + githubAutomatic: { context: { ...review.event, deliveryId: item.receipt }, revision: review.configurationRevision, policy: review.policySnapshot } } }))); + await db.insert(chatActions).values(extra.map(item => ({ ...action, id: randomUUID(), providerActionId: "github-check:" + item.receipt, + payload: { ...action.payload, event: { ...review.event, deliveryId: item.receipt } } }))); + await db.insert(chatGitHubReviews).values(extra.map(item => ({ ...review, id: randomUUID(), deliveryId: item.id }))); + await db.insert(chatGitHubReviews).values(extra.map(item => ({ ...review, id: randomUUID(), deliveryId: "run:" + item.id + ":" + review.headSha }))); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + try { + await registerGitHubReviewProjectionWork(db, coordinator, () => true).ready; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + const rows = await readChatGitHubReviewsByEndpointId(f.endpoint.id); + expect(rows.filter(row => row.state === "superseded")).toHaveLength(106); + expect(rows.filter(row => row.state === "queued")).toHaveLength(105); + expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + 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 abbfc3b469f..6c95d795541 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1,3 +1,4 @@ +import { registerGitHubReviewProjectionWork } from "./services/chat-channels/github/review-projection-work.js"; import { finishServiceShutdown } from "./shutdown.js"; import { promptInjectionPreventionRoutes } from "./routes/prompt-injection-prevention.js"; import { voiceSessionRoutes, voiceWebhookRoutes } from "./routes/voice-sessions.js"; @@ -199,11 +200,7 @@ import { nativeChatProgressWork } from "./services/chat-native-progress-work.js" import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js"; import { chatFailedRetryNoticeWork } from "./services/chat-failed-retry-notice-work.js"; import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js"; -import { - createCoalescedAsyncTrigger, - isChatPublicationCommitSignal, -} from "./services/chat-publication-reconciliation.js"; -import { subscribeAllCompanyLiveEvents } from "./services/live-events.js"; +import { createCoalescedAsyncTrigger } from "./services/chat-publication-reconciliation.js"; import { heartbeatService } from "./services/heartbeat.js"; import { pluginLifecycleManager } from "./services/plugin-lifecycle.js"; import { registerPluginLifecycleDelivery } from "./services/plugin-lifecycle-delivery.js"; @@ -1289,6 +1286,7 @@ export async function createApp( chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); + registerGitHubReviewProjectionWork(db, deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerGitHubReviewCheckWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerGitHubTerminalResponseWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); @@ -1301,29 +1299,17 @@ export async function createApp( deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(), ); emailChannels.start(); - const reconcileChatPublicationMaintenance = async () => { - await chatChannels.processPublicationMaintenance(); - }; const chatReconciliation = createChatReconciliationCoordinator({ // Chat delivery, action, and Telegram discovery workers now own these // duties. Keep the coordinator's other maintenance lanes independent. processPendingDeliveries: async () => undefined, processFailedGitHubWebhookDeliveries: () => chatChannels.processFailedGitHubWebhookDeliveries(), - flushPublications: () => reconcileChatPublicationMaintenance(), + flushPublications: async () => undefined, onError: (lane, err) => { logger.error({ err, lane }, `Failed to reconcile chat ${lane}`); }, }); - const unsubscribeChatPublicationSignals = subscribeAllCompanyLiveEvents( - (event) => { - if (!isWarmStandby() && !isIdleTaskDrainActive() && isChatPublicationCommitSignal(event)) - chatReconciliation.notifyPublications(); - }, - ); - // Remaining publication maintenance still uses this cadence. Lifecycle and - // native progress projection have independent commit/deadline workers. - // Inbox, outbox, and file receipt queues have independent commit/deadline wakes. let chatPublicationTimer: ReturnType | null = setInterval( () => { if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile(); @@ -1458,7 +1444,6 @@ export async function createApp( jobCoordinator.stop(); resourceLifecycleDelivery.dispose(); await deliveryWork.stop(); - unsubscribeChatPublicationSignals(); chatReconciliation.stop(); if (chatPublicationTimer) { clearInterval(chatPublicationTimer); diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index d4ec773f63f..06e74126247 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -13,7 +13,6 @@ import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messag import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; import { resolveInboundConversationBinding } from "./chat-channels/inbound/conversation-binding.js"; import { githubResponseCommentService } from "./chat-github-response-comments.js"; -import { createGitHubMaintenance } from "./chat-channels/github/maintenance.js"; import { createSpekoChannelService, authorizeSpekoPublicVoiceTask, replaceSpekoChannelRuntime, SPEKO_CONVERSATION_IS_LINEAR } from "./chat-channels/speko/runtime.js"; import { prepareDiscordRuntimeCommand } from "./chat-channels/discord/runtime-commands.js"; import { createSlackSetupServices } from "./chat-channels/slack/setup-services.js"; @@ -3346,18 +3345,14 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { } const githubReviews = githubChatReviewService(db, fetchImpl); - const { scheduleGitHubMaintenance } = createGitHubMaintenance(db, fetchImpl, { - scheduleMessageProcessing, isShuttingDown: () => shuttingDown, - }); - - async function processPublicationMaintenance(limit = 25) { - scheduleGitHubMaintenance(); + async function processPublicationMaintenance() { + await githubReviewCheckService(db, fetchImpl).repairSupersededReviews(); } // The app refills free endpoint slots on commits and durable deadlines. // Work remains tracked by this service and is joined before runtime shutdown. async function schedulePendingPublications(limit = 25) { - await processPublicationMaintenance(limit); + await processPublicationMaintenance(); return processPendingPublications(limit, { waitForCompletion: false, maintenance: false }); } diff --git a/server/src/services/chat-channels/github/maintenance.ts b/server/src/services/chat-channels/github/maintenance.ts deleted file mode 100644 index 29394a24814..00000000000 --- a/server/src/services/chat-channels/github/maintenance.ts +++ /dev/null @@ -1,23 +0,0 @@ -import type { Db } from "@paperclipai/db"; -import { githubReviewCheckService } from "../../chat-github-checks.js"; - -export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fetch, options: { - scheduleMessageProcessing: (task: () => Promise) => void; - isShuttingDown: () => boolean; -}) { - let githubMaintenancePending = false; - function scheduleGitHubMaintenance() { - if (githubMaintenancePending || options.isShuttingDown()) return; - githubMaintenancePending = true; - options.scheduleMessageProcessing(async () => { - try { - // Provider I/O must not block Slack, Teams, or ordinary message receipts. - await githubReviewCheckService(db, fetchImpl).repairSupersededReviews(); - } finally { - githubMaintenancePending = false; - } - }); - } - - return { scheduleGitHubMaintenance }; -} diff --git a/server/src/services/chat-channels/github/review-projection-work.ts b/server/src/services/chat-channels/github/review-projection-work.ts new file mode 100644 index 00000000000..9da2da809ec --- /dev/null +++ b/server/src/services/chat-channels/github/review-projection-work.ts @@ -0,0 +1,75 @@ +import { and, asc, eq, gt, isNull, sql } from "drizzle-orm"; +import { chatActions, chatDeliveries, chatGitHubReviews, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js"; + +type Scope = { companyId: string; endpointId: string }; +type Entry = Scope & { id: string; dueAt: number }; + +export async function notifyGitHubReviewProjection(tx: object, scope: Scope) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubReviewProjection, JSON.stringify(scope)); +} + +function pendingProjection() { + return and( + sql`${chatGitHubReviews.state} in ('queued', 'running')`, + isNull(chatGitHubReviews.assessment), + sql`${chatGitHubReviews.deliveryId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'`, + sql`( + select action.id from ${chatActions} action + where action.company_id = ${chatGitHubReviews.companyId} + and action.endpoint_id = ${chatGitHubReviews.endpointId} + and action.kind = 'github_review_check' and action.status = 'cancelled' + and action.result->>'code' = 'stale_head' + and action.payload->'event'->>'deliveryId' = ( + select delivery.normalized_event->'githubAutomatic'->'context'->>'deliveryId' + from ${chatDeliveries} delivery + where delivery.id = case + when ${chatGitHubReviews.deliveryId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' + then ${chatGitHubReviews.deliveryId}::uuid end + and delivery.endpoint_id = ${chatGitHubReviews.endpointId} + ) + limit 1 + ) is not null`, + ); +} + +export async function repairSupersededGitHubReviews(db: Db, id?: string) { + await db.update(chatGitHubReviews).set({ state: "superseded", updatedAt: new Date() }) + .where(and(id ? eq(chatGitHubReviews.id, id) : undefined, pendingProjection())); +} + +export function registerGitHubReviewProjectionWork( + db: Db, + coordinator: ReturnType, + canRun: () => boolean, +) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.chatGitHubReviewProjection, + decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: Scope) => !scope || + (entry.companyId === scope.companyId && entry.endpointId === scope.endpointId), + load: async (scope?: Scope, signal?: AbortSignal) => { + const entries: Entry[] = []; + let cursor: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await db.select({ id: chatGitHubReviews.id, companyId: chatGitHubReviews.companyId, + endpointId: chatGitHubReviews.endpointId }).from(chatGitHubReviews).where(and(pendingProjection(), + scope ? eq(chatGitHubReviews.companyId, scope.companyId) : undefined, + scope ? eq(chatGitHubReviews.endpointId, scope.endpointId) : undefined, + cursor ? gt(chatGitHubReviews.id, cursor) : undefined, + )).orderBy(asc(chatGitHubReviews.id)).limit(100); + entries.push(...rows.map(row => ({ ...row, dueAt: 0 }))); + if (rows.length < 100) return entries; + cursor = rows[rows.length - 1].id; + } + throw signal?.reason ?? new Error("GitHub review projection recovery stopped"); + }, + check: async (entry: Entry, signal?: AbortSignal) => { + if (!canRun() || signal?.aborted) return entry; + await repairSupersededGitHubReviews(db, entry.id); + return 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 6d5988d1990..5b8cfb450bc 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 { notifyGitHubReviewProjection } from "./review-projection-work.js"; import { notifyGitHubReviewChecks } from "./review-check-work.js"; import { agents, chatDeliveries, chatGitHubReviews } from "@paperclipai/db"; import { and, eq, sql } from "drizzle-orm"; @@ -44,6 +45,7 @@ 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 notifyGitHubReviewProjection(taskTx, { companyId: endpoint.companyId, endpointId: endpoint.id }); 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 c9fe43534c7..63e1f9321f8 100644 --- a/server/src/services/chat-github-checks.ts +++ b/server/src/services/chat-github-checks.ts @@ -1,3 +1,4 @@ +import { notifyGitHubReviewProjection, repairSupersededGitHubReviews } from "./chat-channels/github/review-projection-work.js"; 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"; @@ -5,10 +6,9 @@ import { GitHubPublicationLeaseLost, withGitHubPublicationLease, } from "./chat-github-publication-lease.js"; -import { and, eq, inArray, isNull, or, sql } from "drizzle-orm"; +import { and, eq, isNull, or, sql } from "drizzle-orm"; import { chatActions, - chatDeliveries, chatEndpoints, chatGitHubConfigurations, chatGitHubReviews, @@ -163,6 +163,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { ); }); await lease.commit(async (tx) => { + await notifyGitHubReviewProjection(tx, { companyId: action.companyId, endpointId: action.endpointId }); await tx .update(chatActions) .set({ @@ -226,6 +227,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { ); if (review) await lease.commit(async (tx) => { + await notifyGitHubReviewProjection(tx, { companyId: action.companyId, endpointId: action.endpointId }); await tx .update(chatGitHubReviews) .set({ @@ -285,42 +287,7 @@ export function githubReviewCheckService(db: Db, fetchImpl = fetch) { }, ); } - 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 - // starting work or borrowing another delivery's authority. - await db - .update(chatGitHubReviews) - .set({ - state: "superseded", - updatedAt: new Date(), - }) - .where( - and( - inArray(chatGitHubReviews.state, ["queued", "running"]), - isNull(chatGitHubReviews.assessment), - sql`${chatGitHubReviews.deliveryId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'`, - sql`( - select action.id from ${chatActions} action - where action.company_id = ${chatGitHubReviews.companyId} - and action.endpoint_id = ${chatGitHubReviews.endpointId} - and action.kind = 'github_review_check' - and action.status = 'cancelled' - and action.result->>'code' = 'stale_head' - and action.payload->'event'->>'deliveryId' = ( - select delivery.normalized_event->'githubAutomatic'->'context'->>'deliveryId' - from ${chatDeliveries} delivery - where delivery.id = case - when ${chatGitHubReviews.deliveryId} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' - then ${chatGitHubReviews.deliveryId}::uuid end - and delivery.endpoint_id = ${chatGitHubReviews.endpointId} - ) - limit 1 - ) is not null`, - ), - ); - } + const repairSupersededReviews = () => repairSupersededGitHubReviews(db); async function processPending(limit = 10) { await repairSupersededReviews(); const rows = await db diff --git a/server/src/services/chat-github-reviews.ts b/server/src/services/chat-github-reviews.ts index cdddc6b3eae..99c139d95ee 100644 --- a/server/src/services/chat-github-reviews.ts +++ b/server/src/services/chat-github-reviews.ts @@ -1,3 +1,4 @@ +import { notifyGitHubReviewProjection } from "./chat-channels/github/review-projection-work.js"; import { notifyGitHubReviewChecks } from "./chat-channels/github/review-check-work.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { githubRunReplyState } from "./chat-run-publications.js"; @@ -352,6 +353,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) { if (prior && prior.runId === source.run.id) return prior; if (prior && !prior.runId && prior.deliveryId === source.delivery.id) { const [claimed] = await db.transaction(async tx => { + await notifyGitHubReviewProjection(tx, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id }); await notifyGitHubReviewChecks(tx, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id, repositoryId: source.repositoryId, number: source.number }); return tx diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index afe718a27ec..4651bf27b31 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", + chatGitHubReviewProjection: "chat-github-review-projection", chatGitHubReviewChecks: "chat-github-review-checks", chatGitHubTerminalResponses: "chat-github-terminal-responses", chatGitHubAcknowledgements: "chat-github-acknowledgements", diff --git a/tests/e2e/agent-configuration.spec.ts b/tests/e2e/agent-configuration.spec.ts index d09b17dc864..fce0a04a067 100644 --- a/tests/e2e/agent-configuration.spec.ts +++ b/tests/e2e/agent-configuration.spec.ts @@ -45,10 +45,15 @@ const editor = (page: Page) => page.getByRole("textbox", { name: "Agent YAML con async function document(page: Page): Promise { await expect(editor(page)).toBeVisible(); await expect(editor(page)).toHaveAttribute("contenteditable", "true"); - await editor(page).press("ControlOrMeta+a"); - await editor(page).press("ControlOrMeta+c"); - const source = await page.evaluate(() => navigator.clipboard.readText()); - expect(source).toContain("apiVersion: paperclip/v1"); + let source = ""; + await expect.poll(async () => { + await page.evaluate(() => navigator.clipboard.writeText("")); + await editor(page).click(); + await editor(page).press("ControlOrMeta+a"); + await editor(page).press("ControlOrMeta+c"); + source = await page.evaluate(() => navigator.clipboard.readText()); + return /^apiVersion: paperclip\/v1(?:\r?\n|$)/.test(source); + }).toBe(true); return parse(source); } async function edit(page: Page, change: (value: AgentConfigurationDocument) => void) {