fix(server): wake stale GitHub review repair from committed changes (#17664)

<!-- entire-trail-link-start -->
https://entire.com/gh/paperclipai/-/trails/215/changes/paperclip/215
<!-- entire-trail-link-end -->

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - GitHub checks and task admission can finish in either order.
> - A check can reject an old head before admission creates its review
row.
> - The chat maintenance loop repeatedly scans for that late review.
> - Existing cancellation and delivery receipts already identify the
repair.
> - This pull request wakes repair from committed changes and leaves
settled history quiet.

## Linked Issues or Issue Description

**What happened?**

GitHub chat maintenance repeatedly scans all review projections. Empty
instances and retained history incur database work even when no repair
is owed.

**What did you expect?**

A committed stale-head cancellation or admitted review should wake
repair. Only the exact delivery's unassessed queued/running review
should become superseded. When no repair remains, this duty should stop
making database calls.

**Steps to reproduce**

Configure a GitHub chat endpoint and leave it idle after all reviews
settle. The previous maintenance callback continues to call
repairSupersededReviews. A rapid push can also cancel a check before
admission inserts its review, which explains why recovery must remain.

Refs #17662, the merged review-check publication parent. Related #17661
and #17656 migrate terminal responses and acknowledgements. This slice
owns only late stale-head projection repair.

## What Changed

- Move the repair predicate and execution into the GitHub provider
owner. Match the original admitted delivery and cancellation receipt.
Recheck state and assessment in the update.
- Register a scoped worker with the existing coordinator. Use keyset
pages and fair bounded dispatch. Keep retries only for unresolved repair
or failed reconstruction.
- Signal before cancellation, automatic receipt insertion,
automatic-review binding, and check-driven review-state writes. Preserve
existing transaction ownership and outer commit behavior.
- Remove the periodic provider wrapper and publication-maintenance
callback. Preserve explicit manual recovery and independent
missed-webhook recovery.
- Add two concurrent, idempotent partial indexes. Use a guarded delivery
UUID lookup and a bounded exact-receipt lookup. Add no tables.
- Add ten real-database lifecycle and producer cases. Document the
producer audit, failure history, and remaining limits.

- Keep YAML browser reads scoped to the editor after focus changes, with
the original journey assertions and deadlines.

## Verification

Current source: `0886a0229b1c4f68d2f582da86aa1f248e9124f3`.

- At `dd6bca0485c3444d6e3d0ee10ec4aa651e801392`, **287 distinct cases
pass**: 274 complete GitHub workflow cases and 13 startup cases. The
GitHub command filters out 1,485 unrelated cases. The workflow suite
uses a new, pre-migrated PostgreSQL database.
- Coverage includes both orders of cancellation and admission, outer
commit and rollback, restart, drain, standby, failed-update retry,
concurrent assessment under an observed row lock, shutdown during that
wait, more than one candidate page, manual history, and settled-history
quiet.
- Database build, server typecheck/build, module boundaries, and target
migration order pass.
- Canonical migration `0403_natural_cerise.sql` preserves both original
index definitions and concurrent idempotent creation. All 401 parent
migration receipts remain unchanged during upgrade. One new receipt is
added. Replay adds none. Both indexes are ready and valid on fresh and
upgraded databases. The five newest snapshots remain.
- The normal parent merge preserved the prepared tree. The later browser
fixture repair changes only its central read helper and verification
note. Entire server, package, and UI trees remain identical to the
accepted source.
- [The owning
audit](https://github.com/paperclipai/paperclip/blob/0886a0229b1c4f68d2f582da86aa1f248e9124f3/doc/plans/2026-10-25-github-review-projection-sleep-audit.md)
records the exact evidence and earlier failures. Earlier query-plan
measurements retain their historical revision scope.
- Fresh current-head CI and Greptile review remain required. No deployed
sleep, full repository-suite, or fleet-savings claim is made.

Failure accounting: browser shard 7 failed twice on
`1453a4447c0d4d33f713ae3e00734c7950f95426` because the YAML helper read
whole-page clipboard text while main was focused. Three isolated
unchanged local passes did not resolve that failure. A controlled native
focus-loss probe reproduced the same error before the repair and passed
after it. The helper now clears stale clipboard data, focuses the
editor, and waits within the existing expectation budget for a fresh
YAML copy. All six complete YAML journeys pass with zero retries in 50.9
seconds. Journey assertions and deadlines are unchanged. The temporary
fault is removed. The exact CI focus-change trigger remains unknown.
[Both CI
attempts](https://github.com/paperclipai/paperclip/actions/runs/40777754306)
remain part of the verification record. Fresh new-head checks are
required.

Ponytail follow-up: the 4/5 review of `b3c81ba` was conditional on
fixing a stale startup-test slice marker, so the merge was held.
Boundary assertions reproduced the invalid -1 end index. The test now
uses the existing timer declaration and checks both bounds; all 13
startup/hostname cases pass. The optional no-op coordinator cleanup is
already in #17666. The six YAML journeys and all production code remain
unchanged. Fresh review and CI are required for this corrected head.

## Risks

- Missing a future producer signal could delay discovery until restart.
Supported writers share the current database owner's process-local
notification scope. Independent SQL writers and other server processes
need their own integration.
- The migration builds indexes concurrently. The existing migration
runner handles interruption and invalid index recovery.
- Direct GitHub missed-webhook discovery remains periodic. This change
does not make the whole chat service idle.
- Existing Cloud wake coverage and sleep blockers remain unchanged.
These tests do not prove deployed host or database suspension.
- Each remaining sleep slice needs its own review and CI checks before
merge.

## Model Used

OpenAI Codex (GPT-6), with reasoning and code execution. The exact
serving model ID and context-window size were not exposed.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 18:42:54 -05:00
1 parent c2163b3ecf
commit 8bc051245f
19 files changed
+735 -101

No files matched your search

@@ -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.
@@ -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.
+1
View File
@@ -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",
@@ -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}$';
@@ -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": [
+7
View File
@@ -2808,6 +2808,13 @@
"when": 1793223470886,
"tag": "0402_ancient_malice",
"breakpoints": true
},
{
"idx": 403,
"version": "7",
"when": 1793224779622,
"tag": "0403_natural_cerise",
"breakpoints": true
}
]
}
+3
View File
@@ -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'
+3
View File
@@ -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],
@@ -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");
@@ -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<void>(resolve => { entered = resolve; });
const hold = new Promise<void>(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<unknown> | 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<Parameters<Db["transaction"]>[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<unknown> | 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<void>(resolve => { entered = resolve; });
const wait = new Promise<void>(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<typeof review.assessment> = { 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<void>(resolve => { locked = resolve; }), held = new Promise<void>(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();
+4 -19
View File
@@ -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<typeof setInterval> | 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);
+3 -8
View File
@@ -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 });
}
@@ -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>) => 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 };
}
@@ -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<typeof createDeliveryWorkCoordinator>,
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);
}
@@ -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<void> {
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,
+5 -38
View File
@@ -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
@@ -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
@@ -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",
+9 -4
View File
@@ -45,10 +45,15 @@ const editor = (page: Page) => page.getByRole("textbox", { name: "Agent YAML con
async function document(page: Page): Promise<AgentConfigurationDocument> {
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) {