fix(server): wake GitHub acknowledgements from committed changes (#17656)

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

## Thinking Path

> - Paperclip coordinates agent work and presents its progress in chat.
> - GitHub deliveries save acknowledgement actions and provider
receipts.
> - Periodic maintenance currently searches for unfinished
acknowledgements even when idle.
> - Committed delivery, lease, response and presentation changes
identify when work can advance.
> - The existing owner should wake on those changes and recover only at
saved deadlines.
> - Completed history should become quiet without losing unfinished
provider effects.

## Linked Issues or Issue Description

Refs #17653, which is merged. This existing GitHub acknowledgement slice
targets master.

**Problem:** A saved GitHub acknowledgement can wait for a publication
lease, fast-response result or native presentation decision. Polling
empty or unchanged work keeps the tenant active. Removing maintenance
without covering these producers can strand the acknowledgement.

**Expected behavior:** Committed changes wake the scoped owner. It
reconstructs existing actions on startup, retries at saved deadlines,
parks unchanged presentation waits and becomes quiet when settled.

## What Changed

- Register GitHub acknowledgement work with the shared scoped deadline
coordinator. Reuse existing action, delivery, lease and fast-response
rows.
- Add transaction-aware response-availability notifications at supported
writes. Keep GitHub protocol details in its platform owner and shared
notifications limited to IDs.
- Share the bounded latest-request/run projection between scheduling and
direct response authorization. An absent native presentation decision
parks the action until its real writer commits.
- Retain publication leases, source/principal authority, version
fencing, response markers and final receipt precedence. Check admission
before each new provider request; let in-flight work settle. Stop does
not consume the provider retry budget.
- Remove only recurring acknowledgement recovery and keep its manual
path. Add no work table or scheduler service.
- Regenerate the three original indexes as
`0400_absent_phantom_reporter.sql`: pending acknowledgement IDs,
fast-response delivery lookup and publication comment/destination
lookup. Each uses concurrent/idempotent creation and the existing
top-level migration runner. Preserve all landed SQL/journal entries and
the five newest snapshots.

## Verification

Published head: `09ca00b9f029838434ec042cda691345d67b493e`, on landed
continuation parent `063cfd43813cabf18417605fb9eefb9aacf95c8d`.

- **87 distinct current cases pass:** 18 real PostgreSQL acknowledgement
lifecycle/presentation cases, all 26 fast-response service cases, all 30
heartbeat completion cases and all 13 startup cases. Database build,
server typechecking/build, module boundaries and three fresh migrations
through 0400 pass. The current Runner bundle rebuild also passes after
upstream provider-event changes.
- The lifecycle matrix covers empty/history quiet, lease release,
restart, drain/standby, fast-response failure/expiry/publication/unknown
outcome, streaming quarantine, outer rollback/commit, saved provider
retry, stop between requests and receipt state. The real presentation
writer proves quiet while pending, rollback, committed reactivation and
final quiet.
- The earlier **305-case** matrix (236 full GitHub workflow cases, 26
fast-response service cases, 30 completion cases and 13 startup cases)
passes at `337f272853d4f8a51b5b9300900a46eb4e13f698`. It retains its
prepared-parent scope. Repeated cases are not added to current 87.
- A populated 0399 database upgrades to 0400 with all 398 prior
migration receipts unchanged, one new receipt and no additional receipt
on replay. All three indexes are valid/ready on fresh and upgraded
databases. Their SQL definitions match the original draft after
whitespace normalization; migration history and retained snapshot bytes
are preserved.
- Normal integration `9b3607e159153e00ccdfe538d52b591bd38d6499`
preserves the full 28-path owned binary patch and all 20 unrelated
master paths, including native retry, provider-event and evaluation
workflow changes. Both normal merge parents and the exact tree are
checked. No force push or rebase occurs.
- [The owner
audit](https://github.com/paperclipai/paperclip/blob/09ca00b9f029838434ec042cda691345d67b493e/doc/plans/2026-10-25-github-chat-maintenance-sleep-audit.md#acceptance-on-the-actual-landed-continuation-parent)
retains the original missing-notification, JSON precedence, receipt
lifecycle, external-database fixture and migration-registry failures.
Earlier 69/228/broad counts keep their original scope. Skips and failed
revisions are not relabelled as current passes.
- The final audit-only edit leaves all owned source/test/migration
hashes unchanged. **Fresh exact-head CI and Greptile 5/5 with no
actionable findings remain required.** No full local repository, live
GitHub tenant or deployed sleep result is claimed.

## Risks

- Process-local transaction hints cover supported application writers
and shared pools. Independent SQL writers and other replicas do not gain
a new wake guarantee.
- Provider effects remain guarded by the existing publication lease and
current source/principal authority. Unknown outcomes retain their
existing recovery semantics.
- Presentation waits depend on their supported milestone/transaction
producers. A missing producer can defer recovery until another supported
change or restart.
- Index creation can take time on large histories and runs concurrently
outside a transaction. Existing migration recovery handles replay; prior
SQL is unchanged.
- Worker quiet is distinct from host/database suspension. Cloud wake
coverage and sleep admission remain unchanged; later maintenance duties
are separate slices.

## Model Used

OpenAI Codex, using reasoning, repository inspection, code execution and
real PostgreSQL tests. The exact model identifier and context-window
size are not exposed in this session.

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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 16:06:42 -05:00
1 parent 063cfd4381
commit ba5bba86dc
28 files changed
+1253 -111

No files matched your search

@@ -2326,3 +2326,46 @@ start another send after drain/standby/shutdown. Three regressions reproduced
it. Per-send admission/abort checks now pass all 14 new lifecycle cases, and the
previously skipped publication suite plus startup guards pass 27 cases with the
external test database. Fresh corrected-head verification remains required.
## Slice 33 accepted verification and GitHub owner split
#17653 at `276ef23d2a0b01b15e6bfb8ff31ab13495741b19` passes all 53
remote checks, fresh Greptile 5/5, and has no unresolved review threads. Workspace
typecheck/build pass. All 1,540 tests across chat integration, standalone
interaction publication, and startup guards passed on a fresh database with no
skips in 473.47 seconds. The initial skipped suite and the stop-boundary failure
remain documented. The PR is unmerged.
Slice 34 starts from that verified head and owns initial GitHub response-comment
acknowledgement delivery. The same maintenance helper also contains terminal
response reconciliation, review-check publication, and late stale-head review
projection repair. Those are separate reasons and remain separate planned PRs.
See `2026-10-25-github-chat-maintenance-sleep-audit.md` for the inspected owners,
producer boundaries, and response-slot dependencies. No GitHub migration is yet
claimed implemented or verified.
## Slice 34: initial GitHub acknowledgement delivery
Based on accepted slice 33 (`276ef23d2a0b01b15e6bfb8ff31ab13495741b19`).
The implementation moves only initial response-comment acknowledgement out of
recurring GitHub maintenance. The existing response actions, fast-response
records, and publication leases own durability. Committed response-slot changes
and actual saved deadlines drive recovery. Terminal responses, review checks,
and stale-head projection remain separate upcoming reasons.
Fourteen new real-PostgreSQL lifecycle cases and the existing startup guard pass.
The producer-removal regression fails as expected when fast-response settlement
has no notification. Migration application/replay pass after adding the generated
concurrent-index migration to the existing top-level execution registry. Broader
local and remote acceptance is pending. See the detailed
[GitHub owner audit](2026-10-25-github-chat-maintenance-sleep-audit.md).
All earlier failing CI, review findings, stack/master reconciliation, and remaining
inventory duties retain their existing requirements. This slice does not claim
that the stack is ready or authorize a merge.
Slice-34 final focused acceptance passes 17 new cases plus the startup guard.
Workspace typecheck/build pass again after the receipt-lifecycle correction.
The fast-response suites pass 69 tests. The initial complete chat run passed
1,554 tests; a fresh complete run with the three added lifecycle regressions
is in progress. No remote gate or full-stack readiness claim is made yet.
@@ -0,0 +1,409 @@
# Remaining GitHub chat maintenance owners
Inspected after slice 33 at `6d01f4c730aed91e660a63eede688f4f4da61252`.
No implementation or verification claim yet.
The label "GitHub history maintenance" hides several separate duties.
`chat-channels/github/maintenance.ts` serially calls response-comment
`processPending`, then review-check `processPending`. It launches through the
chat service's tracked message-processing lifecycle, with a single-flight flag.
Do not replace that tracked work with an unjoined promise or shared-provider code.
## Candidate slice 34: response-comment acknowledgement dispatch
Owner: `chat-github-response-comments.ts`. Existing durable actions have
kind `github_response_comment`. `stageResponse` inserts status received before
competing for the repository publication lease, so contention does not lose work.
It is called both by acknowledgement and `writeGitHubResponseComment`, including
agent-authored/final response paths. Insertion currently uses a root client;
notification must be in a transaction around the actual write. Repeated calls
also read the existing action, and must not repeatedly wake themselves solely
because an insert lost onConflictDoNothing.
The first half of `processPending` selects received actions and failed actions
with result.retryable=true whose saved retryAt is due, oldest-created first,
limit 10. `tryAcknowledge` records attempts and exponential retryAt capped at
five minutes; existing policy stops automatic retries after eight attempts.
403 authorization changes cancel the action. Preserve that current policy,
company/endpoint/principal authority, generation fencing, and lease ownership.
Do not infer successful delivery from a missing row or an SDK return alone.
Acknowledgement can deliberately defer to a fast response while its request or
publication is pending/running/streaming, or until its saved expiresAt. Published
or ambiguous fast responses supersede the ordinary acknowledgement without
another send. A worker cannot simply repeatedly run every received action with
now+backoff: park behind committed fast-response/publication state and retain
only the real expiration/recovery obligation. Inspect all those producers.
Provider history scanning is bounded at 100 pages of 100 comments and fails
closed if identity cannot safely be recovered. Preserve App-owned markers,
shared pull-response targets, legacy placeholder retirement, no-regression final
receipts, and uncertain-post handling. Provider details stay in this owner.
## Separate reason: terminal response reconciliation
The second half of response `processPending` scans processed actions with
result.final != true. It joins conversation/endpoint/issue and latest terminal
run for the assigned agent, with blocked-issue or source-message provenance
conditions. Acknowledgement completion is not final-response completion.
Run transitions, issue changes, source linkage, response/publication settlement,
and fast-response changes can affect eligibility. Current milestone notifications
may provide an existing boundary but require a full writer audit.
`acknowledge` preserves source-run/active-identity checks and defers while
`githubRunReplyState` is unsettled. A successful final reply may require retiring
an older placeholder rather than posting a second reply. Existing no-op maintenance
writes updatedAt only to rotate oldest-prefix candidates; removing that polling
must replace starvation handling without manufacturing permanent writes.
Waiting for a nonterminal run is passive, while unresolved actual delivery still
needs owned recovery. Keep this independently reviewable from initial dispatch.
## Separate reason: GitHub review-check publication
Owner: `chat-github-checks.ts`, kind `github_review_check`. Enqueue comes from
webhook-events and chat-github-reviews. It snapshots runtime generation and
normalized review event. Delivery derives queued/running/completed state from
review, run, inbound delivery, assessment and current configuration. Existing
repository publication leases and GitHub head/authority checks remain mandatory.
A terminal assessment has its own outbox; ordinary check delivery marks itself
processed when that assessment owns the result. Never dispatch agents here.
Nonterminal success currently saves received plus retryAt=now+5s, even if
nothing changed. Replace that state discovery with producer events. Failed
publication has saved exponential retryAt and the existing eight-attempt policy.
Investigate stale-head discovery: an external PR push must have a supported
webhook/replay contract or a scoped outstanding-check reconciliation deadline.
Do not remove its only fallback on assumption that local run events cover it.
## Separate reason: late review projection repair
Before dispatch, check processPending globally updates queued/running review
rows without assessments when a matching cancelled stale-head check receipt
exists. This repairs admission that commits after cancellation. Both writers
must signal it; late review insert and cancellation are separate commits.
Extract this duty from publication dispatch rather than keep a global update
on every check wake. Preserve exact company/endpoint/delivery/event identity.
All four duties still need real producer tests, startup reconstruction, scoped
fair progress, rollback/commit, drain/resume, and quiet retained-history tests.
The aggregate maintenance call cannot be retired until every duty is covered.
## Additional producer/dependency evidence
Response writes also enter from `chat-github-reviews.ts` (agent comment/update
and assessment paths), `chat-channels/publications/transport.ts` (failed-run
publication), and `github/wake-authority.ts` after inbound admission. The
`tryAcknowledge` boundary classifies current authority failures as cancellation;
it does not silently return null for a missing source. Preserve that distinction.
Fast-response insertion already signals `DELIVERY_QUEUES.fastResponse` through
`fast-responses.ts`, but that notification currently has no scope and later
request-state transitions need their own audit. Publication notifications carry
optional issue scope only on relevant writers; not every publication mutation
currently fans out to milestones. A future acknowledgement worker must cover
both request and publication transitions without importing GitHub policy into
these shared services. The concrete cross-owner capability is notification of
a change to an existing response slot, not a GitHub-specific scheduler branch.
Existing integration fixtures and regression tests are in the early GitHub
section of `chat-channels.integration.test.ts` (around lines 2,750-6,200 in this
revision); there are no standalone response-comment or check suites at the
initial guessed filenames. Reuse the controlled provider API and real DB
fixtures rather than replace authority/lease operations with mocks.
## Design constraints refined during slice 33 acceptance
Keep initial acknowledgement dispatch separate from terminal-response recovery.
The public/manual `processPending` can delegate to both, while recurring
maintenance calls only the terminal half until its own later migration.
The repository publication lease already releases through a transaction and
signals `chatGitHubPublications`. A new acknowledgement worker must not register
a second worker under that occupied topic. It needs its own topic and narrowly
scoped lease-release hints or a shared response-availability notification.
Lost/contended lease acquisition returns undefined; that is not proof of a
successful acknowledgement. Represent the actual outstanding lease expiry and
react to release. Renewals are active-operation work, not idle polling.
Avoid a self-wake loop: `stageResponse` is called repeatedly while acknowledging
an existing action. A no-op insert/read must not repeatedly trigger dispatch of
that same received action. A received action waiting on fast-response publication
needs either the saved fast-response expiry, publication outcome changes, or
concrete unresolved claim recovery. Never return now indefinitely for it.
Most ordinary publication terminal writers already supply company/issue scope to
`notifyChatPublicationWork`; provider retry/unknown and stale-stream quarantine
paths sometimes omit it. Specifically inspect `publications/failures.ts`,
`worker.ts::settlePublicationPreparationFailure`, and queue quarantine. A fast
response that changes streaming -> retry/unknown must release/reclassify the
acknowledgement's wait. Do not depend solely on succeeded publication events.
Fast-response state writers include enqueue, skip, settle, claim, publication
comment binding, supersede, final publication-status suppression, and bulk expiry.
Only changes to the actual acknowledgement predicate require a wake. Expiry is
already a saved deadline; claim pending -> running keeps the same predicate.
Early failure/skip and adding a publication binding require producer coverage.
Do not blindly wrap every write or introduce GitHub branches in fast-responses.
Use a normalized response-slot change contract if these shared owners need a
new notification capability, with existing transactions reused wherever present.
## Slice 34 implementation and boundaries
Initial response acknowledgement now registers with the existing delivery
coordinator. This is a separate worker from terminal-response reconciliation,
review-check delivery, and stale-head repair. The latter three remain in the
GitHub maintenance owner. The explicit/manual combined `processPending` entry
point still performs both initial and terminal response recovery.
The durable source remains `chat_actions.kind = github_response_comment`.
Received actions and retryable failed actions are reconstructed on startup in
100-row ID pages. The shared scoped-deadline helper dispatches a fair batch of
25 across companies. Exact delivery, conversation, issue, and repository-lease
hints restrict reconstruction. Processed/cancelled actions and exhausted failures
leave this worker's inventory; retained response history does not arm a timer.
No new queue table or changed automatic retry limit is introduced.
The next deadline is derived from existing saved retry state, a competing
repository publication lease, or a pending fast response's expiration. A
streaming fast response parks acknowledgement work until publication settlement
or quarantine changes its state. The publication worker still owns stale-stream
recovery. This worker does not invent a second timeout or replay an uncertain
publication. Published or delivery-unknown fast responses suppress the ordinary
acknowledgement through the existing provider owner.
Producer coverage:
- `stageResponse` signals its insert transaction. Existing response lookups do
not emit another signal. This prevents acknowledgement itself from continuously
waking an unchanged received action.
- Failed/cancelled acknowledgement settlement signals the existing action's
delivery before its transaction writes the result.
- Repository publication lease release signals the precise endpoint/repository/
pull-or-issue number. The signal covers receipt changes made under that lease,
and releases waiting acknowledgements when another publication finishes.
- Fast-response admission, early skip, accounting settlement, and publication
binding signal the existing response slot. Pending-to-running claims retain
the same wait predicate; pending expiry already has a saved deadline.
- `supersedeFastResponse` is used only by Board chat, which has no external
acknowledgement delivery. It was audited and left unchanged.
- Publication completion, failure, retry, ambiguous delivery, preparation failure,
pause-race reset, and stale-stream quarantine notify response availability.
Task-control cancellation covers the affected conversation. Content preparation
and lease bookkeeping do not create response-availability debt.
- Endpoint removal deletes its response actions. Other authority checks remain
in the response service before any provider write. The worker does not cache
authorization or infer authority from a queued action.
The shared capability is `notifyChatResponseAvailability`: an identifier-only
hint that an existing response slot may have changed. Fast-response and ordinary
publication owners do not inspect GitHub payloads or choose GitHub behavior.
Only the GitHub acknowledgement owner parses repository thread identity and
applies its response protocol. Notification serialization explicitly picks IDs
so passing a publication row cannot carry its payload into scheduler hints.
Provider requests remain outside database transactions. The existing repository
lease, current source/principal checks, response markers, shared pull-response
identity, version fencing, and final-receipt preservation remain in charge of
provider effects. A recovered known receipt is marked processed so it cannot
remain permanently in the pending acknowledgement inventory. Admission is checked
at the actual fetch boundary after lease renewal; an in-flight request may settle,
but a stopped worker cannot start the next provider request. Stop-related failures
do not consume the provider retry budget.
The migration adds three concurrent, idempotent indexes: pending acknowledgement
IDs, fast-response delivery lookup, and publication comment/destination lookup.
It uses the existing top-level migration mechanism and snapshot pruning rules.
These indexes avoid rebuilding pending work by scanning completed history.
Supported writes are the service's process-local transaction notification model.
Independent SQL writers or a second instance do not acquire a new wake guarantee
from this change. Host sleep eligibility and external alarm coverage are unchanged.
A local deadline is not evidence that a stopped host can wake itself.
## Verification evidence retained during development
- The first eight lifecycle tests failed on a PostgreSQL JSON-expression precedence
error in the lease lookup. Parenthesizing the repository-ID expression fixed
the query; all eight then passed.
- Removing the real fast-response settlement notification made the regression
fail: fallback waited until the response expired. Restoring that signal is the
implementation under test. The assertion measures when the action settles,
separately from the coordinator's final coalesced bookkeeping pass.
- The initial concurrent-index application failed because the migration was not
in the existing top-level statement registry. Registering it fixed application;
a second application also completed successfully.
- Expanded tests exposed provider error normalization at the fetch boundary.
The worker now preserves a stopped attempt without charging a provider failure.
The fault fixture was also narrowed to response-comment POSTs so installation
access-token requests were not mistaken for acknowledgement attempts.
- Fourteen new real-PostgreSQL cases and the startup guard pass. They cover empty
and retained-history quiet, lease release, restart, drain, standby, fast-response
failure/expiry/publication/unknown outcome, stale-stream quarantine, outer
rollback/commit, saved provider retry, and stop between provider requests.
- Full chat/fast-response acceptance, workspace typecheck/build, remote CI and
review remain pending. This document does not yet claim PR readiness.
Workspace typecheck/build now pass. The fast-response test harness supports an
explicit fresh `_test` PostgreSQL database, retaining embedded PostgreSQL by
default. Its first run passed the 69 test bodies but failed cleanup on an invalid
client-close call; the corrected setup then hit retained fixed-prefix fixtures
when the same database was reused. A fresh database passes all 69 tests and cleanup.
Final server typecheck passes. All three migration indexes are valid in PostgreSQL.
The complete affected chat suites are still running; remote checks/review have not
been claimed complete.
A rollback-only PostgreSQL plan experiment seeded 10,000 completed response
actions and 10,000 unrelated fast-response records alongside one pending action
and its matching request. The pending-ID lookup used
`chat_actions_github_ack_work_idx` and read one row (two shared buffers); the
request lookup used `fast_response_requests_delivery_idx` and read one row
(three shared buffers). This checks the two lookup access paths, not whole-worker
latency or deployed fleet savings. All experimental writes were rolled back.
The first complete chat acceptance passed all 1,554 tests without skips in
496.42 seconds. Final review then added three receipt-lifecycle cases: confirmed
received/failed receipts settle, but cancelled receipts stay cancelled. The new
cancelled case failed before correction. Settlement now applies only to pending
states and compares the observed status before writing. The final version of the
producer-removal test also failed as expected without the settlement notification.
The corrected implementation is undergoing a fresh full acceptance run with the
three additional cases; earlier full results are not substituted for that run.
After the final receipt correction, all 17 new lifecycle cases and the startup
guard pass. Workspace typecheck and build pass again. The final complete chat
run remains in progress; source and branch remain fixed during acceptance.
## Acknowledgement presentation wait correction
An acknowledgement for the latest successful native run can remain `received`
while that run's presentation decision is still absent. Reattempting it only
acquires/releases the publication lease and returns, leaving an overdue entry.
Recovery now shares the delivery owner's request/run projection and parks that
specific no-reply state. It retains normal lease, fast-response, and delivery
retry deadlines for actionable requests. Run/presentation/publication changes
wake it through the existing transaction-aware response-availability topic.
No new rows or periodic recovery scan are introduced.
The GitHub owner keeps request binding and provider reply interpretation in
`github/response-run-state.ts`; both delivery and scheduling use the same
projection. Heartbeat's existing presentation write is named
`recordRunPresentationDecision` without changing its transaction or event order,
so the real producer can be exercised with outer commit and rollback. This does
not move GitHub publication authority into the generic native chat repair path.
The first usable regression reproduced an overdue deadline in
`/tmp/sleep-ack-presentation-red-terminal.log`. Earlier fixture attempts lacked
GitHub configuration or terminal success. An attempted generic native-response
repair correctly returned false: GitHub replies remain tool-owned. Those fixture
failures remain in the red/configured, green, and current logs. The final regression
uses the existing GitHub delivery fixture and real heartbeat decision writer:
no SQL for two virtual minutes while parked or after rollback; commit updates the
same provider comment once, then returns to quiet.
All 228 GitHub workflow cases pass (`/tmp/sleep-ack-presentation-acceptance.log`),
as do all 30 heartbeat completion cases (`/tmp/sleep-ack-completion-full.log`) and
server typecheck (`/tmp/sleep-ack-presentation-types-final.log`). The initial
completion command omitted its database URL and skipped 29 cases; the full result
uses a fresh dedicated migrated database. No full-repository, final-stack, or
deployed sleep claim follows from these owner-level checks.
## Current parent and migration acceptance — October 28
The existing GitHub acknowledgement slice is normally integrated onto published
continuation parent `a1d16f84b3fae361d5d648670a5964486370c3a3`. Integration
`51f2f35f54d75648e50f9ec9beada1e479307ecb` preserves every original domain/schema
edit and every unrelated parent blob. It carries the parent confirmation and
continuation work indexes. Proof:
`/tmp/sleep34-indexed-parent-composition-proof.json`.
The old draft-owned migration 0394 collided with landed history. Its three
indexes were first regenerated locally as 0399; the preceding continuation then
needed its own proven work index at 0399. This slice now generates
`0400_absent_phantom_reporter.sql` against the current schema. Only unlanded
artifacts were replaced. The three original concurrent/idempotent SQL index
definitions are identical after whitespace normalization. All prior SQL files,
journal entries and retained snapshot bytes remain intact. Canonical generation
prunes 0395 and keeps the five newest snapshots, 0396–0400. The top-level
migration registry includes the new filename. Both normal merge parents and the
exact tree are checked; no force push or rebase occurs.
At `337f272853d4f8a51b5b9300900a46eb4e13f698`, **305 distinct cases pass**:
- All 236 selected GitHub agent-review workflow cases, with 1,485 other chat
cases skipped. This includes acknowledgement lease/restart/drain/standby,
fast-response outcomes, outer rollback/commit, saved retry, stop between sends,
receipt-state preservation and the real heartbeat presentation writer.
- All 26 cases in the complete fast-response service suite.
- All 30 cases in the complete heartbeat completion suite, with an explicit
external database so no embedded-platform guard skips the cases.
- All 13 app/startup cases.
Database build, server typechecking/build, module boundaries, three fresh
PostgreSQL migrations through 0400 and the exact parent-to-head migration-order
gate pass. The first migration-order invocation passed symbolic `HEAD` to a
command that requires a full 40-character SHA; its usage output is retained.
The corrected command passes without a code change.
An existing populated 0399 test database also upgrades to 0400: all 398 prior
migration receipts remain identical, exactly one receipt is added, and replay
adds none. All three indexes are valid and ready in both a fresh and upgraded
database. Receipt: `/tmp/sleep34-upgrade-replay-receipt.json`. This tests schema
application and replay; it does not claim whole-worker latency or fleet savings.
Commands and exact fixtures: `/tmp/sleep34-current-acceptance-command.json`.
Logs/exit receipts: `/tmp/sleep34-current-{db-build,types,build,boundaries,github,fast,completion,startup}.*`
and `/tmp/sleep34-current-migrate-{github,fast,completion}.*`. The earlier 69-case
fast-response group, 228-case GitHub group and broad chat runs above retain their
original revision scopes. In particular, the current complete service file has
26 cases; an earlier broader count is not substituted for it.
This remains local preparation while #17653 finishes current-head remote gates.
Its eventual landed squash parent must be reconciled before publication. The
following acceptance-only edit leaves all owned source/test/migration hashes
unchanged. Fresh exact-head CI and Greptile 5/5 with no actionable findings are
still required for this slice. No full local repository suite, live GitHub tenant
or deployed sleep result is inferred from this acceptance.
## Acceptance on the actual landed continuation parent
Continuation recovery #17653 has merged at
`063cfd43813cabf18417605fb9eefb9aacf95c8d`. Normal two-parent integration
`9b3607e159153e00ccdfe538d52b591bd38d6499` preserves the full 28-path owned
binary patch and all 20 intervening master paths. Those paths include native
retry recovery, OpenCode provider-event exit codes and evaluation report-lock
workflow fixes. The verified published parent is the semantic base for historic
squash conflicts. Both normal parents and the exact tree are checked; there is
no force push or rebase. Proof:
`/tmp/sleep34-landed-parent-publication-proof.json`.
**87 distinct current cases pass on that landed-parent source:** all 18 selected
acknowledgement lifecycle/presentation cases (1,703 other chat cases skipped),
all 26 fast-response service cases, all 30 heartbeat completion cases and all 13
startup cases. Database build, server typechecking/build, module boundaries,
three fresh PostgreSQL migrations through 0400 and the current-master migration
order gate pass. The Runner vendor bundle is rebuilt successfully because its
upstream provider-event inputs changed. Database build runs first so the prior
local terminal-response preparation's ignored 0401 output cannot leak into this
0400 acceptance.
This selection rechecks the supported signals, claims, rollback/commit,
presentation parking, stop boundaries and settlement consumers after the master
context changes. The earlier 305-case matrix remains scoped to prepared revision
`337f272853d4f8a51b5b9300900a46eb4e13f698`; the repeated 87 cases are not added to
305. All three index definitions and schema history are unchanged during this
integration. The populated upgrade/replay and fresh/upgrade index-validity proof
above therefore keeps its stated schema scope.
Commands, revision and actual fixtures:
`/tmp/sleep34-landed-acceptance-command.json`. Receipts/logs:
`/tmp/sleep34-landed-{db-build,types,build,boundaries,github,fast,completion,startup}.*`,
`/tmp/sleep34-landed-migrate-{github,fast,completion}.*`, and
`/tmp/sleep34-landed-migration-order.*`.
The following audit-only commit leaves all owned source/test/migration hashes
unchanged, recorded in `/tmp/sleep34-landed-tested-source-sha256.json`.
Original failures and earlier scopes above remain visible. Publication requires
fresh exact-head CI and Greptile 5/5 with no actionable findings. No full local
repository, live GitHub tenant or deployed sleep result is inferred.
+1
View File
@@ -551,6 +551,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
"0397_known_madrox.sql",
"0398_yielding_random.sql",
"0399_ambitious_onslaught.sql",
"0400_absent_phantom_reporter.sql",
"0384_funny_adam_warlock.sql",
@@ -0,0 +1,5 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_ack_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_response_comment'
and "chat_actions"."delivery_id" is not null and coalesce("chat_actions"."result"->>'final', 'false') <> 'true'
and ("chat_actions"."status" = 'received' or ("chat_actions"."status" = 'failed' and "chat_actions"."result"->>'retryable' = 'true'));--> statement-breakpoint
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_publications_comment_destination_idx" ON "chat_publications" USING btree ("company_id","comment_id","endpoint_id","conversation_id") WHERE "chat_publications"."comment_id" is not null;--> statement-breakpoint
CREATE INDEX CONCURRENTLY IF NOT EXISTS "fast_response_requests_delivery_idx" ON "fast_response_requests" USING btree ("company_id","endpoint_id","delivery_id") WHERE "fast_response_requests"."delivery_id" is not null;
@@ -1,6 +1,6 @@
{
"id": "66fe02b3-6672-4c19-b447-57acad43bf16",
"prevId": "633afb6b-4b89-44ea-81d5-a477a3696eb5",
"id": "843ab3c7-1378-42f3-b213-401a8d303e3f",
"prevId": "9ca01952-f1fa-4c54-a189-3f5a3a3a3761",
"version": "7",
"dialect": "postgresql",
"tables": {
@@ -9624,6 +9624,50 @@
}
},
"indexes": {
"chat_actions_continuation_work_idx": {
"name": "chat_actions_continuation_work_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'interaction_wakeup' AND \"chat_actions\".\"status\" IN ('issued', 'processing')",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_confirmation_recovery_idx": {
"name": "chat_actions_confirmation_recovery_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'confirmation_response' AND \"chat_actions\".\"status\" = 'processing'",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_provider_action_uq": {
"name": "chat_actions_provider_action_uq",
"columns": [
@@ -9644,6 +9688,22 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_github_ack_work_idx": {
"name": "chat_actions_github_ack_work_idx",
"columns": [
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"delivery_id\" is not null and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -12071,6 +12131,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": [
@@ -14188,6 +14282,24 @@
"primaryKey": false,
"notNull": false
},
"provider_account_id": {
"name": "provider_account_id",
"type": "text",
"primaryKey": false,
"notNull": false
},
"cleanup_confirmed_at": {
"name": "cleanup_confirmed_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"cleanup_next_attempt_at": {
"name": "cleanup_next_attempt_at",
"type": "timestamp with time zone",
"primaryKey": false,
"notNull": false
},
"expires_at": {
"name": "expires_at",
"type": "timestamp with time zone",
@@ -14210,6 +14322,22 @@
}
},
"indexes": {
"chat_voice_inbound_cleanup_idx": {
"name": "chat_voice_inbound_cleanup_idx",
"columns": [
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_voice_inbound_calls\".\"cleanup_confirmed_at\" is null",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_voice_inbound_expiry_idx": {
"name": "chat_voice_inbound_expiry_idx",
"columns": [
@@ -14962,6 +15090,28 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_voice_sessions_report_window_idx": {
"name": "chat_voice_sessions_report_window_idx",
"columns": [
{
"expression": "ended_at",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -30460,6 +30610,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
@@ -2787,6 +2787,13 @@
"when": 1793218102114,
"tag": "0399_ambitious_onslaught",
"breakpoints": true
},
{
"idx": 400,
"version": "7",
"when": 1793218595176,
"tag": "0400_absent_phantom_reporter",
"breakpoints": true
}
]
}
+5
View File
@@ -564,6 +564,8 @@ export const chatPublications = pgTable(
name: "chat_publications_company_comment_fk",
}),
index("chat_publications_work_idx").on(table.state, table.nextAttemptAt),
index("chat_publications_comment_destination_idx").on(table.companyId, table.commentId, table.endpointId, table.conversationId)
.where(sql`${table.commentId} is not null`),
uniqueIndex("chat_publications_idempotency_uq").on(
table.companyId,
table.idempotencyKey,
@@ -683,6 +685,9 @@ export const chatActions = pgTable(
table.endpointId,
table.providerActionId,
),
index("chat_actions_github_ack_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment'
and ${table.deliveryId} is not null and coalesce(${table.result}->>'final', 'false') <> 'true'
and (${table.status} = 'received' or (${table.status} = 'failed' and ${table.result}->>'retryable' = 'true'))`),
foreignKey({
columns: [table.companyId, table.deliveryId],
foreignColumns: [chatDeliveries.companyId, chatDeliveries.id],
+2
View File
@@ -81,6 +81,8 @@ export const fastResponseRequests = pgTable(
t.companyId,
t.sourceKey,
),
delivery: index("fast_response_requests_delivery_idx").on(t.companyId, t.endpointId, t.deliveryId)
.where(sql`${t.deliveryId} is not null`),
history: index("fast_response_requests_history_idx").on(
t.companyId,
t.acceptedAt,
@@ -94,6 +94,10 @@ describe("createChatReconciliationCoordinator", () => {
expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications(");
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).toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()");
expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,");
expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,");
});
@@ -1,3 +1,4 @@
import { recordRunPresentationDecision } from "../services/heartbeat/run-completion.js";
import { environmentService } from "../services/environments.js";
import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js";
@@ -43,7 +44,7 @@ import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-wor
import { aiConnectionService } from "../services/ai-connections.js";
import { fastResponseService } from "../services/fast-responses.js";
import { fastResponseReceipt } from "../services/fast-response-provider.js";
import { fastResponseRequests } from "@paperclipai/db";
import { companyFastResponses, fastResponseRequests } from "@paperclipai/db";
import { toolAccessService } from "../services/tool-access.js";
import { toolActionRequests, toolInvocations, toolPolicies } from "@paperclipai/db";
import { GitHubPublicationLeaseLost, withGitHubPublicationLease } from "../services/chat-github-publication-lease.js";
@@ -4035,6 +4036,240 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
expect(attempts).toBe(2);
expect(f.responseWrites).toHaveLength(writes);
});
it.each(["release", "restart", "drain", "standby"] as const)("wakes GitHub acknowledgements from a real publication lease %s", async mode => {
const f = await reviewBotFixture();
const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id));
const errors: unknown[] = [];
let canRun = true;
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) });
const register = () => f.service.registerGitHubAcknowledgementWork(coordinator, () => canRun);
let worker = register();
let release!: () => void, entered!: () => void;
const hold = new Promise<void>(resolve => { release = resolve; });
const claimed = new Promise<void>(resolve => { entered = resolve; });
let lease: Promise<unknown> | undefined;
try {
await worker.ready;
expect(errors).toEqual([]);
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
expect(coordinator.nextWakeAt()).toBeNull();
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
reads.mockRestore(); writes.mockRestore();
vi.setSystemTime(Date.now() - 120_000);
lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id,
repositoryId: String(resource.metadata.providerRepositoryId), number: 418 }, f.providerFetch, async () => { entered(); await hold; });
await claimed;
if (mode === "restart") await coordinator.stop();
if (mode === "drain" || mode === "standby") canRun = false;
const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) });
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
expect(action.status).toBe("received"); expect(f.responseWrites).toEqual([]);
release(); await lease;
if (mode === "drain" || mode === "standby") {
await vi.advanceTimersByTimeAsync(1000);
expect(f.responseWrites).toEqual([]);
canRun = true; worker.wake();
}
if (mode === "restart") {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
worker = register(); await worker.ready;
}
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(10);
expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].status).toBe("processed");
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
}, { timeout: 10_000 });
expect(f.responseWrites).toHaveLength(1);
expect(f.responseWrites[0].body).toContain("Working on this");
const settledReads = vi.spyOn(db, "select"), settledWrites = vi.spyOn(db, "transaction");
await vi.advanceTimersByTimeAsync(120_000);
expect(settledReads).not.toHaveBeenCalled(); expect(settledWrites).not.toHaveBeenCalled();
settledReads.mockRestore(); settledWrites.mockRestore();
expect(errors).toEqual([]);
} finally {
release(); await lease; await coordinator.stop(); vi.useRealTimers();
}
});
it.each(["failure", "expiry", "published", "delivery_unknown", "streaming quarantine"] as const)("releases a GitHub acknowledgement after fast response %s without maintenance polling", async outcome => {
await instanceSettingsService(db).updateExperimental({ enableFastResponses: true });
const f = await reviewBotFixture();
const binding = await aiConnectionService(db).save(f.companyId, "owner-user", { provider: "openrouter", method: "api_key", name: "Fast GitHub", ownership: "shared", apiKey: "fixture-key", agentIds: [], allAgents: true }, "fixture-key");
const fast = fastResponseService(db, { authorizeExternal: f.service.authorizeFastResponse,
provider: async () => ({ ...(outcome === "failure" ? { errorCode: "invalid_output", noProviderWork: true } : { text: "I’ll check it." }), receipt: fastResponseReceipt({ usage: { inputTokens: 80, outputTokens: 8 } }) }) });
await fast.configure(f.companyId, "owner-user", { enabled: true, ...binding, model: "openai/gpt-oss-120b", allowSponsored: true });
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 f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready;
const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) });
const [job] = await db.select().from(fastResponseRequests).where(eq(fastResponseRequests.companyId, f.companyId));
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(10);
expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime());
}, { timeout: 5000 });
expect(f.responseWrites).toEqual([]);
if (outcome === "expiry") await vi.advanceTimersByTimeAsync(job.expiresAt.getTime() - Date.now());
else {
await fast.process(job);
if (outcome === "streaming quarantine") {
await db.update(chatPublications).set({ state: "streaming", updatedAt: new Date(0) })
.where(eq(chatPublications.idempotencyKey, `fast-response:${job.id}`));
await vi.advanceTimersByTimeAsync(Math.max(0, job.expiresAt.getTime() - Date.now()));
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
expect(f.responseWrites).toEqual([]);
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
reads.mockRestore(); writes.mockRestore();
await f.service.processPendingPublications();
} else if (outcome === "delivery_unknown") {
await db.transaction(async tx => {
await notifyChatPublicationWork(tx, { companyId: job.companyId, issueId: job.issueId });
await tx.update(chatPublications).set({ state: "delivery_unknown" }).where(eq(chatPublications.idempotencyKey, `fast-response:${job.id}`));
});
} else if (outcome === "published") await f.service.processPendingPublications();
}
let processedAt: number | undefined;
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(10);
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
expect(action.status).toBe("processed");
processedAt ??= Date.now();
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
}, { timeout: 10_000 });
if (outcome !== "expiry" && outcome !== "streaming quarantine") expect(processedAt).toBeLessThan(job.expiresAt.getTime());
expect(f.responseWrites).toHaveLength(["failure", "expiry"].includes(outcome) ? 1 : 0);
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
reads.mockRestore(); writes.mockRestore();
expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it("wakes a GitHub acknowledgement only after the outer fast-response transaction commits", async () => {
await instanceSettingsService(db).updateExperimental({ enableFastResponses: true });
const f = await reviewBotFixture();
const binding = await aiConnectionService(db).save(f.companyId, "owner-user", { provider: "openrouter", method: "api_key", name: "Fast GitHub", ownership: "shared", apiKey: "fixture-key", agentIds: [], allAgents: true }, "fixture-key");
const fast = fastResponseService(db, { authorizeExternal: f.service.authorizeFastResponse });
await fast.configure(f.companyId, "owner-user", { enabled: true, ...binding, model: "openai/gpt-oss-120b", allowSponsored: true });
const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) });
const [job] = await db.update(fastResponseRequests).set({ expiresAt: new Date(Date.now() + 60_000) })
.where(eq(fastResponseRequests.companyId, f.companyId)).returning();
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 f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready;
expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime());
const skip = async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
await tx.update(companyFastResponses).set({ enabled: false }).where(eq(companyFastResponses.companyId, f.companyId));
expect(await fastResponseService(tx as unknown as Db).process(job)).toMatchObject({ status: "unavailable" });
await vi.advanceTimersByTimeAsync(100);
expect(f.responseWrites).toEqual([]);
expect((await db.select().from(fastResponseRequests).where(eq(fastResponseRequests.id, job.id)))[0].status).toBe("pending");
};
await expect(db.transaction(async tx => { await skip(tx); throw new Error("rollback fast response skip"); })).rejects.toThrow("rollback fast response skip");
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(idleWorkSnapshot().active).toBe(0); });
expect(f.responseWrites).toEqual([]);
expect(coordinator.nextWakeAt()).toBe(job.expiresAt.getTime());
await db.transaction(skip);
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(10);
expect(f.responseWrites).toHaveLength(1);
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
}, { timeout: 10_000 });
expect(Date.now()).toBeLessThan(job.expiresAt.getTime());
expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it.each(["drain", "standby", "shutdown"] as const)("stops GitHub acknowledgement sends at %s between provider requests", async mode => {
const f = await reviewBotFixture();
f.setSupplementalProviderFetch(async (input, init) => String(input).includes("/issues/418/comments") && init?.method === "POST" ? new Response("unavailable", { status: 503 }) : undefined);
const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) });
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
expect(action.status).toBe("failed"); expect(action.result?.attempts).toBe(1);
await db.update(chatActions).set({ result: { ...action.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, action.id));
let release!: () => void, entered!: () => void;
const held = new Promise<void>(resolve => { release = resolve; });
const started = new Promise<void>(resolve => { entered = resolve; });
f.setSupplementalProviderFetch(async input => {
if (String(input).includes("/issues/418/comments?")) { entered(); await held; }
return undefined;
});
let canRun = true;
const errors: unknown[] = [];
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) });
let worker = f.service.registerGitHubAcknowledgementWork(coordinator, () => canRun);
let stopping: Promise<void> | undefined;
try {
await started;
if (mode === "shutdown") stopping = coordinator.stop(); else canRun = false;
release(); await worker.ready; await stopping;
expect(f.responseWrites).toEqual([]);
expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0].result?.attempts).toBe(1);
canRun = true;
if (mode === "shutdown") {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
worker = f.service.registerGitHubAcknowledgementWork(coordinator, () => true); await worker.ready;
} else worker.wake();
await vi.waitFor(() => { expect(f.responseWrites).toHaveLength(1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 });
expect(errors).toEqual([]);
} finally { release(); await stopping; await coordinator.stop(); }
});
it("retries a failed GitHub acknowledgement at its saved deadline", async () => {
const f = await reviewBotFixture();
let unavailable = true, attempts = 0;
f.setSupplementalProviderFetch(async (input, init) => {
if (String(input).includes("/issues/418/comments") && init?.method === "POST") { attempts++; if (unavailable) return new Response("unavailable", { status: 503 }); }
return undefined;
});
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
try {
await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready;
const thread = makeThread({ channelId: "paperclipai/paperclip", id: "github:paperclipai/paperclip:issue:418" }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41801", text: "Fix this", userId: "42", userName: "octocat", mentioned: true }) });
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
expect(action.status).toBe("failed"); expect(attempts).toBe(1);
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(coordinator.nextWakeAt()).toBe(Date.parse(String(action.result?.retryAt))); });
unavailable = false;
await vi.advanceTimersByTimeAsync(Math.max(0, coordinator.nextWakeAt()! - Date.now()));
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(f.responseWrites).toHaveLength(1); expect(coordinator.nextWakeAt()).toBeNull(); }, { timeout: 10_000 });
expect(attempts).toBe(2);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it.each(["received", "failed", "cancelled"] as const)("preserves the %s GitHub receipt lifecycle during acknowledgement recovery", async status => {
const f = await toolReplyFixture();
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
expect(action.result?.id).toBeTruthy();
await db.update(chatActions).set({ status }).where(eq(chatActions.id, action.id));
const writes = f.responseWrites.length;
await githubResponseCommentService(db, f.providerFetch).tryAcknowledge(action.deliveryId!);
const [settled] = await db.select().from(chatActions).where(eq(chatActions.id, action.id));
expect(settled.status).toBe(status === "cancelled" ? "cancelled" : "processed");
expect(settled.result).toEqual(action.result);
expect(f.responseWrites).toHaveLength(writes);
});
it("does not post a GitHub working comment when the scheduler declines the request", async () => {
const f = await reviewBotFixture();
f.wakeup.mockImplementation(async (agentId, opts) => {
@@ -4167,6 +4402,54 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
expect([...f.responseComments.values()][0].body).toContain("paperclip-response-state:final");
expect(f.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]);
});
it("parks a GitHub acknowledgement until native presentation commits", async () => {
const f = await toolReplyFixture();
const [action] = await db.select().from(chatActions).where(and(
eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
await db.update(chatActions).set({ status: "received", result: null }).where(eq(chatActions.id, action.id));
await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "committed", resultJson: {} })
.where(eq(heartbeatRuns.id, f.run.id));
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
onError: error => errors.push(error) });
const decision = { schema: "paperclip.run_presentation_decision.v1" as const,
commentAction: "none" as const, chosenSource: "none" as const, reasonCodes: ["no_final_response"] };
const commit = (database: typeof db) => recordRunPresentationDecision(database, {
companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!, decision,
});
try {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
await f.service.registerGitHubAcknowledgementWork(coordinator, () => true).ready;
const quiet = async () => {
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10);
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
try {
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
} finally { reads.mockRestore(); writes.mockRestore(); }
};
await quiet();
expect(f.responseWrites).toHaveLength(1);
await expect(db.transaction(async tx => {
await commit(tx as unknown as typeof db);
expect((await readHeartbeatRunsById(f.run.id))[0].resultJson?.presentationDecision).toBeUndefined();
throw new Error("rollback presentation");
})).rejects.toThrow("rollback presentation");
await quiet();
expect(f.responseWrites).toHaveLength(1);
await commit(db);
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(10);
expect((await db.select().from(chatActions).where(eq(chatActions.id, action.id)))[0])
.toMatchObject({ status: "processed", result: { final: true } });
}, { timeout: 10_000 });
expect(f.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]);
expect(f.responseWrites[1].body).toContain("without publishing a final response");
await quiet();
expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it("does not let an unsettled GitHub fallback block a later board message", async () => {
const f = await toolReplyFixture();
await db.insert(chatActions).values({ companyId: f.companyId, endpointId: f.endpoint.id, conversationId: f.conversation.id,
+6 -3
View File
@@ -37,7 +37,7 @@ import { instanceSettingsService } from "../services/instance-settings.js";
import { settleConversationTurn } from "../services/agent-conversations.js";
import { buildLowTrustSourceTrust, LOW_TRUST_QUARANTINED_BODY } from "../services/source-trust.js";
import { deliverBudgetEnforcement } from "../services/budgets.js";
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>,
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined,
db: ReturnType<typeof createDb>,
home: string;
beforeAll(async () => {
@@ -46,12 +46,15 @@ beforeAll(async () => {
);
vi.stubEnv("PAPERCLIP_HOME", home);
vi.stubEnv("PAPERCLIP_INSTANCE_ID", "fast-response-fixture");
database = await startEmbeddedPostgresTestDatabase(
const external = process.env.PAPERCLIP_TEST_DATABASE_URL;
if (external && !new URL(external).pathname.endsWith("_test")) throw new Error("Fast-response tests require a database ending in _test");
database = external ? undefined : await startEmbeddedPostgresTestDatabase(
"paperclip-fast-response-db-",
);
db = createDb(database.connectionString);
db = createDb(external ?? database!.connectionString);
}, 90000);
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await database?.cleanup();
vi.unstubAllEnvs();
if (home) await rm(home, { recursive: true, force: true });
+1
View File
@@ -1289,6 +1289,7 @@ export async function createApp(
chatChannels.registerTeamsConsentExpiryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerTeamsAttemptRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerInteractionContinuationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
+2
View File
@@ -11,6 +11,7 @@ import { credentialFingerprint } from "./chat-channels/credential-fingerprint.js
import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js";
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";
@@ -3503,6 +3504,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
return {
registerInteractionContinuationWork,
registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork,
registerTerminalConfirmationWork,
loadFailedRetryNoticeScopes,
reconcileFailedRetryNotices,
@@ -0,0 +1,112 @@
import { readGitHubResponseRun } from "./response-run-state.js";
import { githubRunReplyState } from "../../chat-run-publications.js";
import { and, asc, eq, gt, isNotNull, or, sql } from "drizzle-orm";
import { chatActions, chatConversations, chatEndpointLeases, chatEndpointResources,
chatPublications, fastResponseRequests, chatEndpoints, chatMessageLinks, type Db } from "@paperclipai/db";
import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js";
import type { ChatResponseScope } from "../../chat-response-availability.js";
import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js";
type Scope = ChatResponseScope & { endpointId?: string; repositoryId?: string; number?: number };
type Entry = { id: string; companyId: string; endpointId: string; deliveryId: string;
issueId: string | null; conversationId: string | null; repositoryId: string | null; number: number | null; dueAt: number };
type FastResponse = { status: string; expiresAt: string; publicationState: string | null };
export function registerGitHubAcknowledgementWork(
db: Db,
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
acknowledge: (deliveryId: string, canContinue: () => boolean) => Promise<unknown>,
canRun: () => boolean,
) {
const repositoryId = sql<string | null>`${chatEndpointResources.metadata}->>'providerRepositoryId'`;
const number = sql<string | null>`substring(${chatConversations.externalThreadId}
from '^github:[^:]+:(?:issue:)?([1-9][0-9]*)(?::rc:[1-9][0-9]*)?$')`;
const selection = {
id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId,
deliveryId: chatActions.deliveryId, conversationId: chatActions.conversationId, issueId: chatConversations.issueId,
agentId: chatEndpoints.assignedAgentId,
sourceMessageId: sql<string | null>`(select m.comment_id from ${chatMessageLinks} m
where m.company_id = ${chatActions.companyId} and m.endpoint_id = ${chatActions.endpointId}
and m.delivery_id = ${chatActions.deliveryId} and m.direction = 'inbound' limit 1)`,
repositoryId, number, status: chatActions.status, result: chatActions.result,
leaseExpiresAt: sql<string | null>`(select max(l.expires_at) from ${chatEndpointLeases} l
where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId}
and l.lease_key = 'github-publication:' || (${repositoryId}) || ':' || ${number})`,
fast: sql<FastResponse | null>`(select jsonb_build_object('status', f.status, 'expiresAt', f.expires_at,
'publicationState', p.state) from ${fastResponseRequests} f left join ${chatPublications} p
on p.company_id = f.company_id and p.comment_id = f.comment_id
and p.endpoint_id = f.endpoint_id and p.conversation_id = f.conversation_id
where f.company_id = ${chatActions.companyId} and f.endpoint_id = ${chatActions.endpointId}
and f.delivery_id = ${chatActions.deliveryId} limit 1)`,
};
async function page(scope?: Scope, cursor?: string, actionId?: string) {
return db.select(selection).from(chatActions)
.innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
.leftJoin(chatConversations, and(eq(chatConversations.id, chatActions.conversationId), eq(chatConversations.companyId, chatActions.companyId)))
.leftJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), eq(chatEndpointResources.companyId, chatActions.companyId)))
.where(and(eq(chatActions.kind, "github_response_comment"), isNotNull(chatActions.deliveryId),
sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`,
or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)),
scope ? eq(chatActions.companyId, scope.companyId) : undefined,
scope?.issueId ? eq(chatConversations.issueId, scope.issueId) : undefined,
scope?.conversationId ? eq(chatActions.conversationId, scope.conversationId) : undefined,
scope?.deliveryId ? eq(chatActions.deliveryId, scope.deliveryId) : undefined,
scope?.endpointId ? eq(chatActions.endpointId, scope.endpointId) : undefined,
scope?.repositoryId ? eq(repositoryId, scope.repositoryId) : undefined,
scope?.number ? eq(number, String(scope.number)) : undefined,
cursor ? gt(chatActions.id, cursor) : undefined,
actionId ? eq(chatActions.id, actionId) : undefined,
)).orderBy(asc(chatActions.id)).limit(100);
}
async function entry(row: Awaited<ReturnType<typeof page>>[number]): Promise<Entry | null> {
const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, {
companyId: row.companyId, endpointId: row.endpointId, agentId: row.agentId,
issueId: row.issueId, deliveryId: row.deliveryId!, sourceMessageId: row.sourceMessageId,
});
// Execution success alone does not decide whether a final response exists.
// Presentation/publication commits invalidate this parked request.
if (row.issueId && ownsLatestRequest && latestRun?.nativePresentationPending &&
await githubRunReplyState(db, { companyId: row.companyId, endpointId: row.endpointId,
issueId: row.issueId, runId: latestRun.id }) === "none") return null;
let dueAt = row.status === "failed" ? Date.parse(String(row.result?.retryAt)) : 0;
if (!Number.isFinite(dueAt)) throw new Error("GitHub acknowledgement has an invalid retry deadline");
const fast = row.fast;
if (fast?.publicationState === "streaming") return null;
if (fast && !["published", "delivery_unknown"].includes(fast.publicationState ?? "") &&
(["pending", "running"].includes(fast.status) || ["pending", "retry"].includes(fast.publicationState ?? ""))) {
dueAt = Math.max(dueAt, Date.parse(fast.expiresAt));
}
if (row.leaseExpiresAt) dueAt = Math.max(dueAt, Date.parse(row.leaseExpiresAt));
return { id: row.id, companyId: row.companyId, endpointId: row.endpointId, deliveryId: row.deliveryId!,
issueId: row.issueId, conversationId: row.conversationId, repositoryId: row.repositoryId, number: row.number ? Number(row.number) : null, dueAt };
}
return registerScopedDeadlineWork(coordinator, {
queue: DELIVERY_QUEUES.chatGitHubAcknowledgements,
decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint),
matches: (value: Entry, scope?: Scope) => !scope || (value.companyId === scope.companyId &&
(!scope.issueId || value.issueId === scope.issueId) && (!scope.deliveryId || value.deliveryId === scope.deliveryId) &&
(!scope.conversationId || value.conversationId === scope.conversationId) &&
(!scope.endpointId || value.endpointId === scope.endpointId) && (!scope.repositoryId || value.repositoryId === scope.repositoryId) &&
(!scope.number || value.number === scope.number)),
load: async (scope?: Scope, signal?: AbortSignal) => {
const entries: Entry[] = [];
let cursor: string | undefined;
while (canRun() && !signal?.aborted) {
const rows = await page(scope, cursor);
for (const row of rows) { const value = await entry(row); if (value) entries.push(value); }
if (rows.length < 100) return entries;
cursor = rows[rows.length - 1].id;
}
throw signal?.reason ?? new Error("GitHub acknowledgement recovery stopped");
},
check: async (value: Entry, signal?: AbortSignal) => {
const [row] = await page(undefined, undefined, value.id);
const current = row ? await entry(row) : null;
if (!current || current.dueAt > Date.now() || !canRun() || signal?.aborted) return current;
await acknowledge(current.deliveryId, () => canRun() && !signal?.aborted);
const [remaining] = await page(undefined, undefined, value.id);
return remaining ? await entry(remaining) : null;
},
}, canRun);
}
@@ -13,7 +13,7 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet
options.scheduleMessageProcessing(async () => {
try {
// Provider I/O must not block Slack, Teams, or ordinary message receipts.
await githubResponseCommentService(db, fetchImpl).processPending();
await githubResponseCommentService(db, fetchImpl).processTerminalResponses();
await githubReviewCheckService(db, fetchImpl).processPending();
} finally {
githubMaintenancePending = false;
@@ -0,0 +1,71 @@
import { and, desc, eq, or, sql } from "drizzle-orm";
import { chatMessageLinks, heartbeatRuns, type Db } from "@paperclipai/db";
import { chatRunSourceMessageId } from "../run-origin.js";
export const responseRunColumns = {
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
responsibleUserId: heartbeatRuns.responsibleUserId,
activeIdentityContextId: heartbeatRuns.activeIdentityContextId,
nativeIssueId: heartbeatRuns.nativeIssueId,
issueId: heartbeatRuns.issueId,
status: heartbeatRuns.status,
updatedAt: heartbeatRuns.updatedAt,
// Terminal execution can precede native presentation commit. Until its
// durable decision exists, missing output is undecided rather than failure.
nativePresentationPending: sql<boolean>`coalesce(
${heartbeatRuns.status} = 'succeeded'
and ${heartbeatRuns.runtimeMode} = 'native'
and (${heartbeatRuns.nativePhase} is distinct from 'committed'
or ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' is distinct from 'paperclip.run_presentation_decision.v1'
or coalesce(${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction', '') not in ('none', 'create', 'reuse')),
false
)`,
// Project only the server-committed no-publication decision, never the
// potentially large native result or model-authored summary.
passiveResponseWait: sql<boolean>`coalesce(
${heartbeatRuns.status} = 'succeeded'
and ${heartbeatRuns.runtimeMode} = 'native'
and ${heartbeatRuns.nativePhase} = 'committed'
and ${heartbeatRuns.resultJson}->>'finalizationPhase' = 'committed'
and ${heartbeatRuns.resultJson}->>'finalizationReasonCode' = 'external_chat_response_waiting'
and ${heartbeatRuns.resultJson}->'nativeResult'->>'schema' = 'paperclip.run_result.v1'
and ${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition' = 'yielded'
and ${heartbeatRuns.resultJson}->'nativeResult'->'continuation'->>'kind' = 'response_wake'
and ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' = 'paperclip.run_presentation_decision.v1'
and ${heartbeatRuns.resultJson}->'presentationDecision'->>'chosenSource' = 'none'
and ${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction' = 'none'
and ${heartbeatRuns.resultJson}->'presentationDecision'->'reasonCodes' @> '["yielded_control_plane_wait"]'::jsonb,
false
)`,
contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object(
'issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
'taskId', ${heartbeatRuns.contextSnapshot}->'taskId',
'wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId',
'wakeCommentIds', ${heartbeatRuns.contextSnapshot}->'wakeCommentIds'
)`,
};
/** Shared request ownership and presentation state for delivery and recovery. */
export async function readGitHubResponseRun(db: Db, input: {
companyId: string; endpointId: string; agentId: string; issueId: string | null;
deliveryId: string; sourceMessageId: string | null;
}) {
const [requestRun] = await db.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).innerJoin(chatMessageLinks, and(
eq(chatMessageLinks.companyId, input.companyId), eq(chatMessageLinks.endpointId, input.endpointId),
eq(chatMessageLinks.deliveryId, input.deliveryId), eq(chatMessageLinks.direction, "inbound"),
or(sql`${chatMessageLinks.commentId}::text = ${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`,
sql`coalesce(${heartbeatRuns.contextSnapshot}->'wakeCommentIds', '[]'::jsonb) ? ${chatMessageLinks.commentId}::text`),
)).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId),
sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${input.issueId}`))
.orderBy(desc(heartbeatRuns.createdAt)).limit(1);
const [latestRun] = await db.select(responseRunColumns).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.agentId, input.agentId),
sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${input.issueId}`,
)).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1);
const originMessageId = latestRun?.activeIdentityContextId ? await chatRunSourceMessageId(db, latestRun,
{ allowLegacyWakeComment: true, provider: "github" }) : null;
const ownsLatestRequest = !!latestRun && (requestRun?.runId === latestRun.id ||
(!!input.sourceMessageId && originMessageId === input.sourceMessageId));
return { latestRun, ownsLatestRequest };
}
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "../../chat-response-availability.js";
import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js";
import { notifyChatConnectionHealth } from "../connection-health-notifications.js";
import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js";
@@ -121,6 +122,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur
? "cancelled"
: "failed";
await notifyChatPublicationWork(tx, state === "failed" || state === "cancelled" ? publication : undefined);
if (state === "retry" || state === "delivery_unknown") await notifyChatResponseAvailability(tx, publication);
const [ownedPublication] = await tx
.update(chatPublications)
.set({
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "../../chat-response-availability.js";
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
import { unresolvedSlackPublicationCondition } from "../../connectors/slack-publication.js";
@@ -242,9 +243,11 @@ export function createChatPublicationQueue(db: Db, options: PublicationQueueOpti
),
),
);
const candidates = await tx.select({ id: chatPublications.id }).from(chatPublications).where(condition).for("update", { skipLocked: true });
const candidates = await tx.select({ id: chatPublications.id, companyId: chatPublications.companyId, issueId: chatPublications.issueId })
.from(chatPublications).where(condition).for("update", { skipLocked: true });
if (!candidates.length) return [];
await notifyChatPublicationWork(tx);
for (const candidate of candidates) await notifyChatResponseAvailability(tx, candidate);
return tx
.update(chatPublications)
.set({
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "../../chat-response-availability.js";
import type { DbOrTransaction } from "../runtime-types.js";
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
@@ -116,6 +117,7 @@ export function createChatPublicationTaskControl(db: Db) {
// finishes after /new or /close. A racing insert that commits after this
// transaction is rejected by the completed-conversation send guard.
await notifyChatPublicationWork(tx);
await notifyChatResponseAvailability(tx, publication);
await tx
.update(chatPublications)
.set({
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "../../chat-response-availability.js";
import { notifyVoicePublication } from "../../voice/voice-reply-notifications.js";
import { createGitHubPublicationPolicy } from "../github/publication-policy.js";
@@ -155,6 +156,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
: Math.min(60_000, 2 ** Math.max(0, attempts) * 1_000);
const result = await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx, terminal ? publication : undefined);
if (!terminal) await notifyChatResponseAvailability(tx, publication);
const [settled] = await tx
.update(chatPublications)
.set({
@@ -475,6 +477,7 @@ export function createChatPublicationWorker(db: Db, options: PublicationWorkerOp
await publicationLease.assertOwned();
await db.transaction(async (tx) => {
await notifyChatPublicationWork(tx);
await notifyChatResponseAvailability(tx, publication);
return tx
.update(chatPublications)
.set({
@@ -116,6 +116,7 @@ export async function withGitHubPublicationLease<T>(
await renewal;
await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify(scope));
await tx.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.id, claimed.id), eq(chatEndpointLeases.token, token)));
});
}
@@ -1,3 +1,4 @@
import { readGitHubResponseRun, responseRunColumns } from "./chat-channels/github/response-run-state.js";
import { and, asc, desc, eq, inArray, or, sql } from "drizzle-orm";
import {
chatActions, chatConversations, chatDeliveries, chatEndpoints,
@@ -5,6 +6,9 @@ import {
fastResponseRequests, chatPublications,
type Db,
} from "@paperclipai/db";
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { registerGitHubAcknowledgementWork } from "./chat-channels/github/acknowledgement-work.js";
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { conflict, forbidden, HttpError } from "../errors.js";
import { githubChatPrincipalAccess } from "./chat-github-access.js";
import { githubAutomaticAdmission } from "./chat-github-events.js";
@@ -25,51 +29,8 @@ type Api = { request<T>(path: string, options?: { method?: string; body?: unknow
type Lease = { commit<T>(write: (tx: Transaction) => Promise<T>): Promise<T> };
type Source = GitHubResponseSource;
type Comment = { id: number; body?: string; html_url: string; user?: { login?: string } };
const responseRunColumns = {
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
responsibleUserId: heartbeatRuns.responsibleUserId,
activeIdentityContextId: heartbeatRuns.activeIdentityContextId,
nativeIssueId: heartbeatRuns.nativeIssueId,
issueId: heartbeatRuns.issueId,
status: heartbeatRuns.status,
updatedAt: heartbeatRuns.updatedAt,
// Terminal execution can precede native presentation commit. Until its
// durable decision exists, missing output is undecided rather than failure.
nativePresentationPending: sql<boolean>`coalesce(
${heartbeatRuns.status} = 'succeeded'
and ${heartbeatRuns.runtimeMode} = 'native'
and (${heartbeatRuns.nativePhase} is distinct from 'committed'
or ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' is distinct from 'paperclip.run_presentation_decision.v1'
or coalesce(${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction', '') not in ('none', 'create', 'reuse')),
false
)`,
// Project only the server-committed no-publication decision, never the
// potentially large native result or model-authored summary.
passiveResponseWait: sql<boolean>`coalesce(
${heartbeatRuns.status} = 'succeeded'
and ${heartbeatRuns.runtimeMode} = 'native'
and ${heartbeatRuns.nativePhase} = 'committed'
and ${heartbeatRuns.resultJson}->>'finalizationPhase' = 'committed'
and ${heartbeatRuns.resultJson}->>'finalizationReasonCode' = 'external_chat_response_waiting'
and ${heartbeatRuns.resultJson}->'nativeResult'->>'schema' = 'paperclip.run_result.v1'
and ${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition' = 'yielded'
and ${heartbeatRuns.resultJson}->'nativeResult'->'continuation'->>'kind' = 'response_wake'
and ${heartbeatRuns.resultJson}->'presentationDecision'->>'schema' = 'paperclip.run_presentation_decision.v1'
and ${heartbeatRuns.resultJson}->'presentationDecision'->>'chosenSource' = 'none'
and ${heartbeatRuns.resultJson}->'presentationDecision'->>'commentAction' = 'none'
and ${heartbeatRuns.resultJson}->'presentationDecision'->'reasonCodes' @> '["yielded_control_plane_wait"]'::jsonb,
false
)`,
contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object(
'issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
'taskId', ${heartbeatRuns.contextSnapshot}->'taskId',
'wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId',
'wakeCommentIds', ${heartbeatRuns.contextSnapshot}->'wakeCommentIds'
)`,
};
const generation = (endpoint: Source["endpoint"]) => (endpoint.setup as { runtimeGeneration?: number }).runtimeGeneration ?? 0;
const stoppedAcknowledgement = Symbol("stopped GitHub acknowledgement");
const actionKey = (deliveryId: string) => `github_response:${deliveryId}`;
const responseMarker = (source: Source) => isMainPullResponse(source) ? pullResponseMarker(source) : requestResponseMarker(source);
@@ -91,18 +52,23 @@ async function findResponseComment(source: Source, api: Api, legacyMarkers: stri
* Reconnect epochs fence publication attempts, not the durable comment identity.
* Legacy receipts are also checked against the App-owned provider comment. */
async function stageResponse(db: Db, source: Source) {
const [inserted] = await db.insert(chatActions).values({
companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
conversationId: source.conversation.id, deliveryId: source.delivery.id,
principalId: source.delivery.principalId, kind: "github_response_comment",
providerActionId: actionKey(source.delivery.id),
payload: { botExternalId: source.endpoint.botExternalId, agentId: source.endpoint.assignedAgentId },
status: "received",
}).onConflictDoNothing().returning();
const response = inserted ?? (await db.select().from(chatActions).where(and(
const lookup = () => db.select().from(chatActions).where(and(
eq(chatActions.companyId, source.endpoint.companyId), eq(chatActions.endpointId, source.endpoint.id),
eq(chatActions.providerActionId, actionKey(source.delivery.id)),
)))[0];
)).then(rows => rows[0]);
const existing = await lookup();
const response = existing ?? await db.transaction(async tx => {
await notifyChatResponseAvailability(tx, { companyId: source.endpoint.companyId, deliveryId: source.delivery.id });
const [inserted] = await tx.insert(chatActions).values({
companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
conversationId: source.conversation.id, deliveryId: source.delivery.id,
principalId: source.delivery.principalId, kind: "github_response_comment",
providerActionId: actionKey(source.delivery.id),
payload: { botExternalId: source.endpoint.botExternalId, agentId: source.endpoint.assignedAgentId },
status: "received",
}).onConflictDoNothing().returning();
return inserted;
}) ?? await lookup();
if (!response || response.conversationId !== source.conversation.id ||
(response.payload.botExternalId !== undefined && response.payload.botExternalId !== source.endpoint.botExternalId) ||
response.payload.agentId !== source.endpoint.assignedAgentId)
@@ -135,6 +101,10 @@ export async function writeGitHubResponseComment(
(input.versionAt && Date.parse(String(response.result.versionAt)) > input.versionAt.getTime()))) {
if (!input.acknowledgement && response.result.final === true && !input.final)
throw conflict("This GitHub response is already final");
if (["received", "failed"].includes(response.status)) await lease.commit(async tx => {
await tx.update(chatActions).set({ status: "processed", updatedAt: new Date() })
.where(and(eq(chatActions.id, response.id), eq(chatActions.status, response.status)));
});
return { id: String(response.result.id), url: String(response.result.url) };
}
const marker = responseMarker(source);
@@ -213,13 +183,18 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
throw forbidden("Invalid GitHub request repository");
return { ...row, repositoryId, repository: match[1]!, number: Number(match[2]), replyId: match[3] ?? null };
}
async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise<void>) {
async function acknowledge(deliveryId: string, finalBody?: string, assertRun?: () => Promise<void>, canContinue = () => true) {
if (!canContinue()) return;
const source = await context(deliveryId);
// Persist before competing for the publication lane. A busy lane leaves
// durable acknowledgement work for maintenance instead of losing it.
await stageResponse(db, source);
if (!canContinue()) return;
return withGitHubPublicationLease(db, { companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
repositoryId: source.repositoryId, number: source.number }, fetchImpl, async lease => {
repositoryId: source.repositoryId, number: source.number }, (input, init) => {
if (!canContinue()) throw stoppedAcknowledgement;
return fetchImpl(input, init);
}, async lease => {
const assertCurrent = async () => {
await assertRun?.();
const current = await context(deliveryId);
@@ -234,22 +209,11 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
let preserveFinal = false;
if (finalBody === undefined) {
const existing = await stageResponse(db, source);
const [requestRun] = await db.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).innerJoin(chatMessageLinks, and(
eq(chatMessageLinks.companyId, source.endpoint.companyId), eq(chatMessageLinks.endpointId, source.endpoint.id),
eq(chatMessageLinks.deliveryId, deliveryId), eq(chatMessageLinks.direction, "inbound"),
or(sql`${chatMessageLinks.commentId}::text = ${heartbeatRuns.contextSnapshot}->>'wakeCommentId'`,
sql`coalesce(${heartbeatRuns.contextSnapshot}->'wakeCommentIds', '[]'::jsonb) ? ${chatMessageLinks.commentId}::text`),
)).where(and(eq(heartbeatRuns.companyId, source.endpoint.companyId), eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId),
sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${source.conversation.issueId}`))
.orderBy(desc(heartbeatRuns.createdAt)).limit(1);
const [latestRun] = await db.select(responseRunColumns).from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, source.endpoint.companyId), eq(heartbeatRuns.agentId, source.endpoint.assignedAgentId),
sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${source.conversation.issueId}`,
)).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1);
const originMessageId = latestRun?.activeIdentityContextId ? await chatRunSourceMessageId(db, latestRun,
{ allowLegacyWakeComment: true, provider: "github" }) : null;
const ownsLatestRequest = !!latestRun && (requestRun?.runId === latestRun.id ||
(!!source.sourceMessageId && originMessageId === source.sourceMessageId));
const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, {
companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
agentId: source.endpoint.assignedAgentId, issueId: source.conversation.issueId,
deliveryId, sourceMessageId: source.sourceMessageId,
});
requestRunning = ownsLatestRequest && latestRun?.status === "running";
const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status);
const run = terminal && (ownsLatestRequest || source.issueStatus === "blocked") ? latestRun : null;
@@ -289,7 +253,10 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
(await stagePullResponse(db, source)).result?.id === existing.result.id)) {
// Rotate no-op maintenance checks so a page of newer queued asks
// cannot hide other conversations that need terminal reconciliation.
await lease.commit(async tx => { await tx.update(chatActions).set({ updatedAt: new Date() }).where(eq(chatActions.id, existing.id)); });
await lease.commit(async tx => { await tx.update(chatActions).set({
status: ["received", "failed"].includes(existing.status) ? "processed" : existing.status,
updatedAt: new Date(),
}).where(and(eq(chatActions.id, existing.id), eq(chatActions.status, existing.status))); });
return { id: String(existing.result.id), url: String(existing.result.url) };
}
if (finalBody === undefined) {
@@ -439,6 +406,9 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
or(eq(chatActions.status, "received"), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`,
sql`(${chatActions.result}->>'retryAt')::timestamptz <= now()`)))).orderBy(asc(chatActions.createdAt)).limit(limit);
for (const row of rows) if (row.deliveryId) await tryAcknowledge(row.deliveryId);
return rows.length + await processTerminalResponses(limit);
}
async function processTerminalResponses(limit = 10) {
// Recovery can stop execution before a milestone publication is staged.
// Revisit acknowledged requests with terminal execution, even after a
// restart; successful acknowledgement is not successful task completion.
@@ -463,21 +433,28 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
or coalesce(r.context_snapshot->'wakeCommentIds', '[]'::jsonb) ? l.comment_id::text))))`,
)).orderBy(asc(chatActions.updatedAt)).limit(limit);
for (const row of stopped) if (row.deliveryId) await tryAcknowledge(row.deliveryId);
return rows.length + stopped.length;
return stopped.length;
}
async function tryAcknowledge(deliveryId: string) {
try { return await acknowledge(deliveryId); }
async function tryAcknowledge(deliveryId: string, canContinue = () => true) {
try { return await acknowledge(deliveryId, undefined, undefined, canContinue); }
catch (error) {
if (error === stoppedAcknowledgement || !canContinue()) return;
const [row] = await db.select().from(chatActions).where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.deliveryId, deliveryId)));
if (!row || row.result?.final === true) return;
const attempts = Number(row.result?.attempts ?? 0) + 1;
const denied = error instanceof Error && "status" in error && error.status === 403;
await db.update(chatActions).set({ status: denied ? "cancelled" : "failed", result: {
...row.result, attempts, retryable: !denied && attempts < 8,
retryAt: new Date(Date.now() + Math.min(300000, 1000 * 2 ** attempts)).toISOString(),
code: denied ? "authorization_changed" : "acknowledgement_unconfirmed",
}, updatedAt: new Date() }).where(and(eq(chatActions.id, row.id), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`));
await db.transaction(async tx => {
await notifyChatResponseAvailability(tx, { companyId: row.companyId, deliveryId });
await tx.update(chatActions).set({ status: denied ? "cancelled" : "failed", result: {
...row.result, attempts, retryable: !denied && attempts < 8,
retryAt: new Date(Date.now() + Math.min(300000, 1000 * 2 ** attempts)).toISOString(),
code: denied ? "authorization_changed" : "acknowledgement_unconfirmed",
}, updatedAt: new Date() }).where(and(eq(chatActions.id, row.id), sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`));
});
}
}
return { tryAcknowledge, processPending, failRun, finishRun, hasWorkingResponseForRun };
return { tryAcknowledge, processPending, processTerminalResponses, failRun, finishRun, hasWorkingResponseForRun,
registerAcknowledgementWork: (coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) =>
registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun),
};
}
@@ -0,0 +1,7 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string };
export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId }));
}
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { notifyChatConfirmationWork } from "./chat-confirmation-notifications.js";
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js";
@@ -15,4 +16,6 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston
// Admission and current chat authority also govern saved queue notices.
await notifyChatInboundWakeNotices(tx, scope);
await notifyChatConfirmationWork(tx, scope);
// Native presentation and run ownership can unblock a parked acknowledgement.
await notifyChatResponseAvailability(tx, scope);
}
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -17,6 +18,7 @@ export async function notifyChatPublicationWork(
// release its provider response slot. Callers supply the affected issue only
// for those changes; transport claim/lease bookkeeping has no projection debt.
if (milestoneScope?.issueId) {
await notifyChatResponseAvailability(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId });
await notifyChatRunMilestones(tx, { companyId: milestoneScope.companyId, issueId: milestoneScope.issueId });
}
}
@@ -41,6 +41,7 @@ export const DELIVERY_QUEUES = {
chatProviderEffects: "chat-provider-effects",
chatGitHubIngress: "chat-github-ingress",
chatGitHubPublications: "chat-github-publications",
chatGitHubAcknowledgements: "chat-github-acknowledgements",
chatFailedRunRetries: "chat-failed-run-retries",
chatTelegramMaintenance: "chat-telegram-maintenance",
chatReceivers: "chat-receivers",
+21 -14
View File
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { insertIssueComment } from "./issue-comment-writes.js";
import { randomUUID } from "node:crypto";
import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js";
@@ -195,6 +196,7 @@ export async function enqueueFastResponse(tx: Db, source: FastResponseSource) {
if (source.acceptedAt.getTime() + FAST_RESPONSE_DEADLINE_MS <= Date.now())
return;
await notifyDeliveryWork(tx, DELIVERY_QUEUES.fastResponse);
if (source.endpointId && source.deliveryId) await notifyChatResponseAvailability(tx, { companyId: source.companyId, deliveryId: source.deliveryId });
await tx
.insert(fastResponseRequests)
.values({
@@ -549,20 +551,23 @@ export function fastResponseService(
}
}
async function skip(request: Request, reason: string) {
await db
.update(fastResponseRequests)
.set({
status: "skipped",
publicationStatus: "suppressed",
errorCode: reason,
finishedAt: new Date(),
})
.where(
and(
eq(fastResponseRequests.id, request.id),
eq(fastResponseRequests.status, "pending"),
),
);
await db.transaction(async tx => {
if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId });
await tx
.update(fastResponseRequests)
.set({
status: "skipped",
publicationStatus: "suppressed",
errorCode: reason,
finishedAt: new Date(),
})
.where(
and(
eq(fastResponseRequests.id, request.id),
eq(fastResponseRequests.status, "pending"),
),
);
});
}
async function settle(request: Request, outcome: FastResponseOutcome) {
await withBudgetSpan("budget.accounting.settlement", () => withAccountingTransaction(
@@ -625,6 +630,7 @@ export function fastResponseService(
},
);
const finishedAt = new Date();
if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId });
await tx
.update(fastResponseRequests)
.set({
@@ -993,6 +999,7 @@ export function fastResponseService(
idempotencyKey: `fast-response:${request.id}`,
payload: safe,
});
if (request.deliveryId) await notifyChatResponseAvailability(tx, { companyId: request.companyId, deliveryId: request.deliveryId });
await tx
.update(fastResponseRequests)
.set({
@@ -346,6 +346,19 @@ function controllerLeaseLossResultJson(signal: AbortSignal) {
: {};
}
/** Persist the resolved presentation and wake dependents in its committing transaction. */
export async function recordRunPresentationDecision(db: Db, input: {
companyId: string; runId: string; issueId?: string; decision: RunPresentationDecision;
}) {
await db.transaction(async tx => {
await notifyChatRunMilestones(tx, { companyId: input.companyId, issueId: input.issueId });
await tx.update(heartbeatRuns).set({
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: input.decision })}::jsonb`,
updatedAt: new Date(),
}).where(eq(heartbeatRuns.id, input.runId));
});
}
export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunCompletionDependencies) {
const {
getRun,
@@ -892,13 +905,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
};
}
await db.transaction(async tx => {
await notifyChatRunMilestones(tx, { companyId: livenessRun.companyId, issueId: issueId ?? undefined });
await tx.update(heartbeatRuns).set({
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision })}::jsonb`,
updatedAt: new Date(),
}).where(eq(heartbeatRuns.id, livenessRun.id));
});
await recordRunPresentationDecision(db, { companyId: livenessRun.companyId,
runId: livenessRun.id, issueId: issueId ?? undefined, decision: presentationDecision });
await appendRunEvent(livenessRun, {
eventType: "run.presentation.resolved",
stream: "system",