fix(server): wake GitHub terminal responses from committed changes (#17661)

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

## Thinking Path

> - Paperclip coordinates agent work and presents results in chat.
> - GitHub acknowledgements already persist the response action and
provider receipt.
> - A terminal run can still owe a final response after acknowledgement.
> - Periodic maintenance searches those rows even when no relevant state
changed.
> - Committed run, issue, presentation, lease and publication changes
identify when recovery can advance.
> - The owner should recover from those events and real deadlines, then
become quiet.

## Linked Issues or Issue Description

Refs #17656, which is merged. This existing terminal-response slice
targets master.

**Problem:** A saved GitHub response can remain nonfinal while its run,
native presentation, publication or source is unresolved. Removing
maintenance without covering those producers loses recovery. Polling
settled history keeps the tenant active.

**Expected behavior:** Committed supported changes wake the terminal
owner. It reconstructs existing actions, preserves provider authority
and pending publication ownership, waits at saved deadlines where
needed, and becomes quiet after settlement.

## What Changed

- Register terminal GitHub response work with the shared scoped deadline
coordinator, using existing action/delivery/publication/run rows.
- Notify response availability from supported run milestones, native
identity, recovery resolution, issue updates and tree/workspace
restoration writes. Preserve outer transaction commit/rollback.
- Keep GitHub response interpretation in its platform owner. Use the
bounded run projection for both direct authorization and scheduling,
including native presentation guards and deliberate passive outcomes.
- Retain the existing publication lease, markers, source/principal
checks and final receipt precedence. A publication owner or uncertain
outcome keeps its existing authority; the worker does not replay it.
- Remove only recurring terminal-response recovery, preserving the
manual path and neighbouring duties. Add no table or scheduler service.
- Regenerate the original sparse terminal-work index as
`0401_yummy_ultimates.sql`, concurrently and idempotently through the
existing top-level migration runner. Preserve all prior SQL/journal
entries and the five newest snapshots.

## Verification

Published head: `88a209e28b969b5229e9a621aeddfd3cbae9b0cd`, integrated
with landed #17656 (`ba5bba86dcdde32b14d4d7dec1b51898e5f6f22d`) and
master `52b86c2032a189b0ea259c5b0cb7158b07ef788c`.

- **74 distinct current cases pass:** 17 terminal-response
lifecycle/producer cases (1,721 other chat cases skipped), all six
queued-comment consumption regressions for the new upstream query
optimization, all 30 heartbeat completion cases, all eight real/unit
issue-tree control cases and all 13 startup cases. Database build,
server typecheck/build, module boundaries and four fresh migrations
through 0401 pass.
- The earlier **304-case** prepared-parent matrix passes at
`aa72f32641b2e2c680342cd49477272349aca9db`: all 253 GitHub workflow
cases, 30 completion cases, eight issue-tree cases and 13 startup cases.
It retains that revision scope. Repeated cases are not added to the
current 74.
- The terminal matrix covers finalized-history quiet, deleted Board
sources, pending presentation/passive outcomes, outer rollback/commit,
paused/new runs, restart/drain/standby, committed native identity, lease
waits, full-page unrelated history, uncertain publication ownership and
tree/workspace/exhausted-recovery producers.
- A populated 0400 database upgrades to 0401 with all 399 prior
migration receipts identical, exactly one added, and none added on
replay. The terminal index is valid/ready on fresh and upgraded
databases. Its SQL matches the original draft after whitespace
normalization; generation keeps snapshots 0397–0401.
- Normal integration `3c8339a541bf68bc9957b9d5a8df80d3a7c3452d`
preserves the complete 24-path owned binary patch and all five new
master paths. The landed parent tree also equals the previously
published parent tree byte for byte. Both normal merge parents and the
exact composition tree are checked. No force push or rebase occurs.
- [The owner
audit](https://github.com/paperclipai/paperclip/blob/88a209e28b969b5229e9a621aeddfd3cbae9b0cd/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md#acceptance-on-the-landed-github-parent-and-current-master)
preserves the original producer, quietness, failure-isolation,
skipped-suite and presentation-wait results with their scopes. Current
passes do not erase earlier failures or turn skips into passes.
- The audit-only follow-up 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 suite, live GitHub
tenant or deployed sleep result is claimed.

## Risks

- Process-local notifications cover supported application writers and
shared pools. Independent SQL writers and other replicas gain no new
wake guarantee.
- Native terminal status alone does not authorize fallback publication.
Pending presentation and existing publication ownership remain separate
obligations.
- Stop can leave an in-flight provider request settling; admission
fences stop a subsequent request. Unknown outcomes retain their existing
recovery rules.
- Index creation can take time on large histories. Concurrent/idempotent
migration and existing replay recovery remain in charge.
- Worker quiet does not establish host/database suspension. Cloud alarm
coverage and sleep admission remain unchanged.

## 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:51:57 -05:00
1 parent f8484a9f8d
commit f8226b0515
24 files changed
+1185 -51

No files matched your search

@@ -2369,3 +2369,32 @@ Workspace typecheck/build pass again after the receipt-lifecycle correction.
The fast-response suites pass 69 tests. The initial complete chat run passed
1,554 tests; a fresh complete run with the three added lifecycle regressions
is in progress. No remote gate or full-stack readiness claim is made yet.
## Slice 34 accepted head and slice 35 checkpoint
[#17656](https://github.com/paperclipai/paperclip/pull/17656) at
`43900c1f94500af4c4de1987f9608ac4e12feae6` passes all 53 remote checks,
fresh Greptile 5/5, and has zero review threads. Final chat/publication/startup
acceptance passes all 1,557 tests with no skips in 514.70 seconds. Workspace
types/build and 69 fast-response cases pass. The PR remains draft and unmerged.
Branch `codex/sleep-35-github-terminal-responses` starts from that head.
It extracts provider-owned terminal-state decisions, records successful paused
presentation versions in existing receipts, and adds a scoped worker plus
committed issue/run/response-slot notifications. Eight new real-database
lifecycle cases and nine existing owner cases pass; server types/build pass.
The broader GitHub workflow suite is running. Startup registration and polling
removal wait for the three identified direct issue writers and remaining audit
and acceptance requirements in the terminal-response audit. No slice-35 PR or
full-stack readiness is claimed yet. Earlier parent CI/review findings, all other
inventory duties, and reconciliation with current master remain required.
Slice 35 now covers the reproduced direct tree-restoration, workspace-admission
and exhausted-recovery producers. Real publication-owner tests preserve both
known-unsent cancellation and ambiguous outcome parking. Startup registers the
new terminal worker and recurring GitHub maintenance no longer invokes that duty.
Fourteen new worker cases plus one existing case pass; workspace types/build and
migration application/replay/safety pass. The partial candidate index excludes
10,000 completed rows in a rollback-only query-plan experiment. Fresh full
chat/publication/startup acceptance is running; remote gates remain required.
See the terminal-response audit for all retained failures and scoped evidence.
@@ -0,0 +1,379 @@
# GitHub terminal-response recovery
Slice 35 starts from acknowledgement PR #17656 at
`43900c1f94500af4c4de1987f9608ac4e12feae6`.
The final parent acceptance passed all 1,557 chat/publication/startup tests without
skips in 514.70 seconds. Parent workspace types/build and 69 fast-response tests
also passed. Greptile reviewed that head at 5/5 without findings. Remote CI was
still completing when this slice started; no merge or full-stack completion is
claimed.
## Obligation and ownership
A received acknowledgement and an owed terminal response are different states
of the existing `github_response_comment` action. Slice 34 owns received actions
and retryable failed deliveries. This slice owns processed, nonfinal actions
whose bound request has a terminal outcome or needs a paused notice. Review-check
publication and stale-head projection repair remain independent duties.
The provider owner must determine exact source/run provenance, latest-run
ownership, and whether a real reply is confirmed or uncertain. Keep those queries
and presentation decisions in `chat-channels/github/response-execution.ts`; shared
notification and scheduler code must not interpret GitHub payloads. Existing
principal/configuration checks, provider markers and repository publication leases
remain the authority for sends and receipt writes.
## Passive history is not unfinished delivery
A paused response deliberately remains nonfinal so a later resumed run can answer
it. The old scan repeatedly revisits it, even when it makes no provider mutation.
The replacement needs a saved paused-presentation version in the existing receipt,
written only with confirmed receipt persistence. A changed run or pause state
invalidates that version. No table or separate queue is needed.
An unrelated newer terminal run does not automatically authorize closing every
older request. Resolve native identity and legacy wake-comment provenance using
the existing owner. Passive non-owning rows must be excluded without acquiring
and releasing a publication lease on every wake. That would otherwise create a
self-wake loop through the release notification.
## Producer audit still required
- Run and issue transitions, identity/lineage changes and restored authority use
the existing milestone notification boundary; extend it once with the narrow
response-state topic after verifying coverage.
- Response action completion and provider publication/assessment settlement must
wake affected terminal candidates. Repository lease release is scoped to its
company, endpoint, repository and PR/issue number.
- An unresolved competing lease owns its saved expiration, not a permanent scan.
- Failed terminal sends remain with the existing acknowledgement retry owner,
including the current attempt budget and ambiguous-outcome policy.
- Startup reconstructs eligible work; drain/standby/shutdown retain skipped work
and prevent starting subsequent provider requests.
## Acceptance required before polling removal
Use real PostgreSQL and the existing controlled GitHub fixture. Cover terminal
run producer commit/rollback, native identity, blocked-but-non-owning requests,
paused-history quiet and resume, final/superseded history, lease release and
restart, uncertain reply settlement, admission changes between provider requests,
changes during a sweep, and candidate prefixes across the bounded page size.
Measure actual SQL and HTTP effects while idle, not only timer registration.
Current state: extracting the provider-owned execution decision and persisting
confirmed paused presentation versions. The terminal worker, complete producer
audit, lifecycle tests, startup wiring and recurring-call removal remain pending.
## Implemented worker checkpoint
The provider execution resolver is extracted. Successful paused receipt writes
save `pausedVersion` in the existing result. The scoped worker reconstructs
processed nonfinal candidates in 100-row ID pages, excludes unrelated run
history and unresolved reply owners, and uses real publication-lease expiration
when another writer owns the lane. A settled paused receipt is passive. New
run state invalidates it. The owning acknowledgement service still checks
current authority and stops subsequent provider requests at admission/shutdown.
Failed sends transfer to the existing acknowledgement retry worker.
The response-availability helper now wakes this worker as well as initial
acknowledgements. The milestone boundary wakes it for existing run, identity,
lineage and authority producers. Issue-service status/assignee changes notify
inside their owning transaction. GitHub publication staging, missing-binding
cancellation and scoped lease release also notify. Registration is exposed by
the provider service for tests, but app startup still does not register it and
the old terminal maintenance call remains until the audit below is complete.
New acceptance covers eight cases: startup terminal recovery and settled-history
quiet; outer issue transaction rollback/commit; paused-history quiet and a later
run; restart; drain; standby; native-identity rollback/commit; real repository
lease contention/release. All eight pass in `/tmp/sleep35-lifecycle-coalesced.log`
(21.88 seconds). The SQL-quiet checks advance 120 seconds and inspect real
`select` and `transaction` calls. GitHub HTTP remains controlled.
Server typecheck and build pass at this checkpoint, with logs
`/tmp/sleep35-final-server-types.log` and `/tmp/sleep35-final-server-build.log`.
All GitHub review-workflow acceptance is running in
`/tmp/sleep35-github-acceptance.log`; no broad result is claimed yet.
Material failures retained:
- `/tmp/sleep35-before-producer.log`: all three first cases failed because the
test coordinator omitted its required `canRun` function. Corrected the fixture.
- `/tmp/sleep35-before-producer-fixed-fixture.log`: startup and paused recovery
passed, while the real issue commit failed to wake terminal response work.
The issue owner now signals the response slot before writing its status.
- `/tmp/sleep35-producer-fixed.log`: the producer regression and 15 adjacent
cases passed; the paused restart assertion observed the legitimate pending
coalesced lease-release pass rather than waiting for settlement.
- `/tmp/sleep35-lifecycle.log`: four passed and four assertions failed at the
default one-second poll timeout or before the coalesced pass. The coordinator
itself coalesces at one second. Explicit five-second observation bounds now
allow that existing behavior; `/tmp/sleep35-lifecycle-coalesced.log` passes all
eight. No production delay or retry policy was widened to make tests pass.
- `/tmp/sleep35-owner-extraction.log`: nine existing ownership/paused/native/
legacy-placeholder cases passed after extraction.
## Direct issue writers found before startup wiring
These are unresolved producer-audit items, not exemptions:
1. `issue-tree-control.ts::restoreIssueStatusesForHold` restores issue status
directly. Its cancellation path already calls `issueService.update`, but the
restoration path does not. Test a restored blocked request with existing
response history and cover its transaction before removing the old sweep.
2. `heartbeat/queue.ts` directly blocks a task whose isolated worktree lacks a
usable project. This can happen before any run is admitted, so a run-terminal
notification cannot cover it. Existing assembled coverage is in
`heartbeat-workspace-branch-containment.test.ts` around the projectless
isolated-worktree rejection case. Verify this producer's response notification.
3. `execution-recovery-resolution.ts::settleUnrecoverableExecutions` can set
status blocked after a run was already terminal. It does not currently send
the response notification. Existing real-database coverage is in
`native-safe-replacement.test.ts` and run-dispatch adapter tests. Preserve its
task/coordinator/run lock order and failure rollback when adding the signal.
The native-finalization reconciler's blocked transition calls issueService and
is covered by the new owning boundary. Recovery/service's inspected blocked
transitions also call issueService. Runner goal statuses are session-goal data,
not issue status, and do not independently change this response predicate.
Additional dynamic status writers, lookup/index cost, uncertain-publication
settlement, page fairness, final startup wiring, full acceptance and remote review
still need completion. Do not present this checkpoint as polling removal.
## Production registration and direct-writer repairs
The three direct writers above now notify response availability before their
writes in the existing transactions. Three real-producer regressions reproduce
missing paused response settlement before the change. Tree cancellation continues
to use issueService; restoration now notifies each restored issue. Queue admission
notifies only a transition into blocked. Exhausted recovery does likewise and
retains its existing lock order and failure behavior.
`/tmp/sleep35-direct-producers-before.log` records the missing tree/recovery wakes.
Its workspace case initially failed earlier because the fixture had not enabled
isolated workspaces, so it exercised normal admission instead of the intended
pre-dispatch refusal. The corrected fixture enables and restores the feature,
drains any executions on cleanup, and reproduces the missing workspace wake in
`/tmp/sleep35-workspace-producer-before.log`. The real issue state reaches blocked
before the missing-notice assertion, so the red result establishes the producer
boundary rather than an unrelated setup failure. All eleven producer/lifecycle
cases pass after repair in `/tmp/sleep35-direct-producers-fixed.log`.
The remaining inspected direct status writes in agent-conversations, Slack
lifecycle/state, access reassignment, checkout/release, and native workspace
reconstruction move away from blocked or remove assignment. They do not create
a new paused-response obligation. Execution eligibility is reread before any
send, so an outstanding lease-delayed candidate observes these changes. New
run/lineage changes still go through the milestone boundary. Generic issue-service
assignment/status mutations notify directly. Agent deletion does not restore an
owed response. Goal status fields are separate session state.
Two further cases drive the real review-publication owner through token acquisition,
a terminal run transition and pre-send authority denial. A known-unsent publication
cancels and wakes terminal fallback through lease release. A saved ambiguous write
remains parked, with no repeated SQL over 120 seconds. The latter does not assert
that the provider write failed or relax host safety. Both pass in
`/tmp/sleep35-publication-settlement.log`. The existing eight-attempt acknowledgement
policy remains the owner of actual failed fallback sends.
The app now registers terminal-response work with warm-standby and idle-drain
gates, and recurring GitHub maintenance no longer calls terminal reconciliation.
The explicit/manual combined recovery entry point is preserved. Review-check
publication and stale-head review repair remain separate duties in maintenance.
## Candidate lookup and final acceptance
Generated migration `0395_jittery_norrin_radd.sql` adds a partial index for
processed, nonfinal response actions. It runs concurrently through the existing
top-level migration registry and is idempotent on replay. Snapshot pruning retains
0387 through 0391, exactly five. Application, replay, numbering/safety checks and
PostgreSQL index validity pass. No table or queue storage was added.
The worker pages candidates by ID, then asks the GitHub owner to resolve actual
run provenance and reply state. This has per-candidate read cost at startup or
an affected scope change; it does not scan while settled. Issue/delivery/repository
hints narrow normal wakes. A 101-row prefix of unrelated retained requests cannot
hide the owed terminal response and does not publish those unrelated requests.
The shared helper provides fair 25-entry dispatch batches after reconstruction.
A rollback-only experiment inserted 10,000 completed actions. The candidate index
query read four eligible rows using an index-only scan and four shared buffers.
See `/tmp/sleep35-index-plan.sql` and `/tmp/sleep35-index-plan.log`. This proves
completed-history exclusion for the candidate access path, not whole-worker latency
or fleet savings. The lifecycle tests prove settled SQL quiet separately.
Verification:
- The earlier extracted-owner GitHub acceptance passed 194 cases in 140.71 seconds
(`/tmp/sleep35-github-acceptance.log`) before the three direct-writer repairs.
- Final focused selection passes 15 cases: fourteen new terminal-worker cases
plus one existing matching case (`/tmp/sleep35-wired-focused.log`, 46.15 seconds).
The startup source suite did not match this focused selector and is included
in the full run below; it is not claimed passed by that focused command.
- Workspace typecheck and build pass after final source/wiring/index changes
(`/tmp/sleep35-workspace-types.log`, `/tmp/sleep35-workspace-build.log`).
- Fresh complete chat/publication/startup acceptance is running on
`paperclip_sleep35_full_20261025_test`, log `/tmp/sleep35-full-chat.log`.
Its expected count is 1,571. Do not claim completion before its terminal result.
- Remote checks/review and any resulting repairs remain required.
Independent SQL writers and multiple replicas gain no new wake protocol. Cloud
alarm coverage and sleep admission are unchanged. The full recurring-server stack,
older failed gates/review findings and final master reconciliation remain open.
## Quietness fixture settlement and isolation (October 26)
The coordinator removes a due timer before it runs reconciliation. Therefore,
`nextWakeAt() === null` alone does not establish that the worker has finished.
The completed-history test now waits for both no scheduled deadline and zero
tracked active work before starting its existing two-minute quiet measurement.
It still requires zero database reads, zero transactions, and no provider writes.
No production worker, deadline, or recovery policy changes in this repair.
The same test previously restored its database spies only after all assertions
passed. A failed quietness assertion left the transaction spy installed. The
later audit fixture bound that existing spy, then replaced its implementation
with a callback through the bound spy, causing recursive calls.
A controlled experiment inserted one real SQL read inside the quiet measurement.
Before cleanup was repaired, the quietness assertion failed and the subsequent
`audit_insert` case failed with a stack overflow instead of its expected audit
error (two failures and one pass). With the same intentional SQL read and
unconditional spy restoration, only the quietness assertion failed; both audit
failure cases passed. The temporary SQL read was removed before acceptance.
This proves failure isolation without weakening the quietness assertion. It does
not independently reproduce the original CI timing race.
All 228 cases in the complete GitHub agent review, Discord automatic command
registration, and resource-change audit groups then passed against a migrated,
disposable PostgreSQL database. Other groups were excluded by the test selector;
this is not a complete chat-file result. Logs:
`/tmp/sleep-chat-quiet-contamination-red.log`,
`/tmp/sleep-chat-quiet-contamination-contained.log`, and
`/tmp/sleep-chat-quiet-owner-tests.log`.
## Master integration: native presentation and passive review waits
Master added two durable presentation guards after this worker was extracted.
The integrated provider resolver retains both through the same bounded run
projection used by direct response authorization. A terminal native status with
an unsettled presentation decision creates no fallback-send obligation yet.
The worker becomes quiet and the existing native finalizer milestone signal
reactivates it after a committed presentation update. A confirmed publication
still takes precedence over an absent fallback.
For a main PR response whose exact owned run deliberately yielded without a
publication, acknowledgement retires its own action under the publication lease
and keeps the shared assessment untouched. Issue-thread and review-thread
fallbacks retain their existing behavior. The cancellation signals response
availability inside the committing transaction. No provider call is replayed.
Acceptance includes master's passive-assessment and malformed/pending-decision
cases plus two worker cases: quiet pending presentation, rolled-back settlement,
committed passive or missing-response settlement, and final drain. The fixture
uses real PostgreSQL with controlled GitHub responses. The new cases must pass
before this integration is published; source merging alone is not acceptance.
## Current local preparation — October 28
Normal integration `ed1c9fcfa10d492bb56ea2e23cd97cddeebe9d5b` adopts the prepared
GitHub acknowledgement parent `3fef8f08c0337301706bc6348a05bb2a2fc50679`.
Every original domain/schema edit and unrelated parent blob is preserved.
The only conflict is a historical snapshot rename/delete; it retains the current
parent's 0396 snapshot byte for byte. The parent migration/client state is
identical before this slice's index is regenerated. Proof:
`/tmp/sleep35-parent-composition-proof.json`.
`706cb6dbf58bf42a5f611319994c2c3907c9a80f` regenerates the original terminal
work index as `0401_yummy_ultimates.sql`. Its concurrent/idempotent index SQL is
identical after whitespace normalization to the old unlanded 0395 definition.
All prior SQL/journal entries and retained snapshot bytes remain unchanged.
Canonical generation keeps the five newest snapshots, 0397–0401. The existing
top-level migration registry uses the new filename.
The exact parent-to-head migration-order gate, database build and production
server typechecking pass on that revision. Commands/receipts:
`/tmp/sleep35-current-compile-command.json` and
`/tmp/sleep35-current-{order,db-build,types}.*`.
These are compilation/history checks, not current lifecycle or PostgreSQL
migration-application acceptance. No current test count is claimed here.
The earlier lifecycle, quietness, provider, failure-isolation and presentation
results above retain their original revision scopes.
This remains local and unpublished while preceding PRs finish their required
gates and merges. Actual landed-parent reconciliation, relevant functional
acceptance, fresh exact-head CI and Greptile 5/5 without actionable findings are
still required. No full local repository, live provider or deployed sleep result
is inferred from this preparation. The following audit-only commit leaves the
compiled source and migration bytes unchanged.
## Current prepared-parent functional acceptance
Normal integration `aa72f32641b2e2c680342cd49477272349aca9db` adopts published
GitHub acknowledgement parent `09ca00b9f029838434ec042cda691345d67b493e`.
The complete original owned binary patch and all inherited master changes are
preserved. Proof: `/tmp/sleep35-published-parent-composition-proof.json`.
**304 distinct cases pass:** all 253 GitHub agent-review workflow cases (1,485
other chat cases skipped), all 30 heartbeat completion cases, all eight cases
across the complete real/unit issue-tree control suites, and all 13 startup
cases. Database build, server typechecking/build, module boundaries and three
fresh PostgreSQL migrations through 0401 pass. The control suite is explicitly
premigrated, preserving its original setup budget. Commands and actual fixtures:
`/tmp/sleep35-current-acceptance-command.json`; receipts/logs:
`/tmp/sleep35-current-{db-build,types,build,boundaries,github,completion,tree,startup}.*`
and `/tmp/sleep35-current-migrate-{github,completion,tree}.*`.
These results retain that prepared-parent revision scope.
## Acceptance on the landed GitHub parent and current master
GitHub acknowledgements #17656 landed at
`ba5bba86dcdde32b14d4d7dec1b51898e5f6f22d`. Its entire tree is byte-identical
to the published parent `09ca00b9f029838434ec042cda691345d67b493e`.
Master then added the queued-comment consumption optimization #18021.
Normal two-parent integration `3c8339a541bf68bc9957b9d5a8df80d3a7c3452d`
adopts master `52b86c2032a189b0ea259c5b0cb7158b07ef788c`. It preserves the full
24-path owned binary patch and all five new master paths. Both normal parents
and the exact tree are checked, including historical squash-ancestry resolution
against the verified published parent. Proof:
`/tmp/sleep35-landed-parent-publication-proof.json`. No force push or rebase occurs.
**74 distinct current cases pass:** all 17 terminal-response lifecycle/producer
cases (1,721 other chat cases skipped), all six queued-comment consumption
regressions for the new upstream optimization, all 30 heartbeat completion cases,
all eight real/unit issue-tree control cases and all 13 startup cases. Database
build, server typechecking/build, module boundaries, four fresh PostgreSQL
migrations through 0401 and the current-master migration-order gate pass.
The repeated 68 cases are not added to the earlier 304-case matrix; its full
GitHub group remains evidence at its stated revision.
The terminal cases cover finalized-history quiet, deleted Board sources, pending
presentation and passive results, outer rollback/commit, paused/new runs,
restart/drain/standby, native identity, lease waits, unrelated full-page history,
uncertain publication ownership and tree/workspace/exhausted-recovery producers.
The six new upstream tests cover bounded repeated reads and preserve saved-comment
JSON, identity, scope, author and cancellation semantics in the execution blocker.
A populated 0400 database upgrades to 0401 with all 399 prior migration receipts
identical and exactly one new receipt. Replay adds none. The terminal index is
valid and ready in both fresh and upgraded databases. The original SQL index
definition is unchanged apart from generation whitespace and filename references.
All prior SQL/journal entries and retained snapshot bytes remain preserved;
canonical retention keeps snapshots 0397–0401. Receipt:
`/tmp/sleep35-upgrade-replay-receipt.json`.
Actual commands, revision and fixtures:
`/tmp/sleep35-landed-acceptance-command.json`. Logs/receipts:
`/tmp/sleep35-landed-{db-build,types,build,boundaries,github,queued,completion,tree,startup}.*`,
`/tmp/sleep35-landed-migrate-{github,queued,completion,tree}.*`,
`/tmp/sleep35-landed-migration-order.*`, and `/tmp/sleep35-upgrade-replay.*`.
The following audit-only follow-up leaves all owned source/test/migration hashes
unchanged, in `/tmp/sleep35-landed-tested-source-sha256.json`.
Original producer, quietness, failure-isolation, skipped-suite and presentation
results above retain their original scopes. Fresh exact-head CI and Greptile 5/5
with no actionable findings remain required. This does not establish a full
local repository result, live GitHub delivery or physical host/database sleep.
+1
View File
@@ -552,6 +552,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
"0398_yielding_random.sql",
"0399_ambitious_onslaught.sql",
"0400_absent_phantom_reporter.sql",
"0401_yummy_ultimates.sql",
"0384_funny_adam_warlock.sql",
@@ -0,0 +1,3 @@
CREATE INDEX CONCURRENTLY IF NOT EXISTS "chat_actions_github_terminal_work_idx" ON "chat_actions" USING btree ("id") WHERE "chat_actions"."kind" = 'github_response_comment'
and "chat_actions"."status" = 'processed' and "chat_actions"."delivery_id" is not null
and coalesce("chat_actions"."result"->>'final', 'false') <> 'true';
@@ -1,6 +1,6 @@
{
"id": "be87a5d8-4164-4d95-89ae-b895701fa21b",
"prevId": "66fe02b3-6672-4c19-b447-57acad43bf16",
"id": "c4c8a780-68b0-4b79-8c3b-5847f506ed7c",
"prevId": "843ab3c7-1378-42f3-b213-401a8d303e3f",
"version": "7",
"dialect": "postgresql",
"tables": {
@@ -9624,6 +9624,50 @@
}
},
"indexes": {
"chat_actions_continuation_work_idx": {
"name": "chat_actions_continuation_work_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'interaction_wakeup' AND \"chat_actions\".\"status\" IN ('issued', 'processing')",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_confirmation_recovery_idx": {
"name": "chat_actions_confirmation_recovery_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'confirmation_response' AND \"chat_actions\".\"status\" = 'processing'",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_provider_action_uq": {
"name": "chat_actions_provider_action_uq",
"columns": [
@@ -9644,6 +9688,38 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_github_terminal_work_idx": {
"name": "chat_actions_github_terminal_work_idx",
"columns": [
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"status\" = 'processed' and \"chat_actions\".\"delivery_id\" is not null\n and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_actions_github_ack_work_idx": {
"name": "chat_actions_github_ack_work_idx",
"columns": [
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_actions\".\"kind\" = 'github_response_comment'\n and \"chat_actions\".\"delivery_id\" is not null and coalesce(\"chat_actions\".\"result\"->>'final', 'false') <> 'true'\n and (\"chat_actions\".\"status\" = 'received' or (\"chat_actions\".\"status\" = 'failed' and \"chat_actions\".\"result\"->>'retryable' = 'true'))",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -12071,6 +12147,40 @@
"method": "btree",
"with": {}
},
"chat_publications_comment_destination_idx": {
"name": "chat_publications_comment_destination_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "comment_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "endpoint_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "conversation_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_publications\".\"comment_id\" is not null",
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_publications_idempotency_uq": {
"name": "chat_publications_idempotency_uq",
"columns": [
@@ -14996,6 +15106,28 @@
"concurrently": false,
"method": "btree",
"with": {}
},
"chat_voice_sessions_report_window_idx": {
"name": "chat_voice_sessions_report_window_idx",
"columns": [
{
"expression": "ended_at",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"chat_voice_sessions\".\"state\" in ('ended', 'failed', 'expired') and \"chat_voice_sessions\".\"provider_session_id\" is not null",
"concurrently": false,
"method": "btree",
"with": {}
}
},
"foreignKeys": {
@@ -30494,6 +30626,34 @@
"method": "btree",
"with": {}
},
"fast_response_requests_delivery_idx": {
"name": "fast_response_requests_delivery_idx",
"columns": [
{
"expression": "company_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "endpoint_id",
"isExpression": false,
"asc": true,
"nulls": "last"
},
{
"expression": "delivery_id",
"isExpression": false,
"asc": true,
"nulls": "last"
}
],
"isUnique": false,
"where": "\"fast_response_requests\".\"delivery_id\" is not null",
"concurrently": false,
"method": "btree",
"with": {}
},
"fast_response_requests_history_idx": {
"name": "fast_response_requests_history_idx",
"columns": [
+7
View File
@@ -2794,6 +2794,13 @@
"when": 1793218595176,
"tag": "0400_absent_phantom_reporter",
"breakpoints": true
},
{
"idx": 401,
"version": "7",
"when": 1793219893186,
"tag": "0401_yummy_ultimates",
"breakpoints": true
}
]
}
+3
View File
@@ -685,6 +685,9 @@ export const chatActions = pgTable(
table.endpointId,
table.providerActionId,
),
index("chat_actions_github_terminal_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment'
and ${table.status} = 'processed' and ${table.deliveryId} is not null
and coalesce(${table.result}->>'final', 'false') <> 'true'`),
index("chat_actions_github_ack_work_idx").on(table.id).where(sql`${table.kind} = 'github_response_comment'
and ${table.deliveryId} is not null and coalesce(${table.result}->>'final', 'false') <> 'true'
and (${table.status} = 'received' or (${table.status} = 'failed' and ${table.result}->>'retryable' = 'true'))`),
@@ -97,7 +97,8 @@ describe("createChatReconciliationCoordinator", () => {
expect(source).toContain("chatChannels.registerGitHubAcknowledgementWork(deliveryWork,");
const github = readFileSync(new URL("../services/chat-channels/github/maintenance.ts", import.meta.url), "utf8");
expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processPending()");
expect(github).toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()");
expect(github).not.toContain("githubResponseCommentService(db, fetchImpl).processTerminalResponses()");
expect(source).toContain("chatChannels.registerGitHubTerminalResponseWork(deliveryWork,");
expect(source).toContain("chatChannels.registerTerminalConfirmationWork(deliveryWork,");
expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,");
});
@@ -1,3 +1,5 @@
import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js";
import { settleUnrecoverableExecutions } from "../services/execution-recovery-resolution.js";
import { recordRunPresentationDecision } from "../services/heartbeat/run-completion.js";
import { environmentService } from "../services/environments.js";
import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
@@ -4036,6 +4038,392 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
expect(attempts).toBe(2);
expect(f.responseWrites).toHaveLength(writes);
});
it("restores terminal GitHub responses and leaves finalized history quiet", async () => {
const f = await toolReplyFixture();
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
const worker = githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true);
try {
await worker.ready;
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }),
{ timeout: 5000 }).toEqual({ next: null, active: 0 });
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
try {
const providerWrites = f.responseWrites.length;
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
expect(f.responseWrites).toHaveLength(providerWrites);
} finally { reads.mockRestore(); writes.mockRestore(); }
expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it("settles a deleted Board-origin request without starving unrelated terminal GitHub replies", async () => {
const invalid = await toolReplyFixture();
const valid = await toolReplyFixture();
const [message] = await db.insert(issueComments).values({ companyId: invalid.companyId,
issueId: invalid.conversation.issueId!, authorType: "user", authorUserId: "owner-user", body: "Board follow-up",
}).returning();
await initializeRunIdentity(db, { companyId: invalid.companyId, runId: invalid.run.id,
issueId: invalid.conversation.issueId!, responsibleUserId: "owner-user", messageIds: [message.id], cause: "instruction" });
await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, message.id));
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() })
.where(inArray(heartbeatRuns.id, [invalid.run.id, valid.run.id]));
const invalidWrites = invalid.responseWrites.length;
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
// The denied request must never reach transport; only the valid request
// uses this provider, even though discovery covers both companies.
const responses = githubResponseCommentService(db, valid.providerFetch);
try {
await responses.registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => [...valid.responseComments.values()][0]?.body, { timeout: 5000 })
.toContain("stopped before completing");
const [denied] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, invalid.endpoint.id),
eq(chatActions.kind, "github_response_comment")));
expect(denied).toMatchObject({ status: "cancelled", result: { retryable: false, code: "authorization_changed" } });
expect(invalid.responseWrites).toHaveLength(invalidWrites);
expect(valid.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]);
await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }),
{ timeout: 5000 }).toEqual({ next: null, active: 0 });
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
try {
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
expect(valid.responseWrites.map(write => write.method)).toEqual(["POST", "PATCH"]);
} finally { reads.mockRestore(); writes.mockRestore(); }
expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it.each(["passive", "missing"] as const)("keeps native presentation idle until its committed GitHub outcome: %s", async outcome => {
const f = await toolReplyFixture("github:paperclipai/paperclip:418");
await db.update(heartbeatRuns).set({ status: "succeeded", nativePhase: "observed", resultJson: null })
.where(eq(heartbeatRuns.id, f.run.id));
const [action] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id),
eq(chatActions.kind, "github_response_comment")));
expect(action.status).toBe("processed");
const baseline = f.responseWrites.length, originalBody = [...f.responseComments.values()][0].body;
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
try {
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }))
.toEqual({ next: null, active: 0 });
expect(f.responseWrites).toHaveLength(baseline);
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
try {
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
expect(f.responseWrites).toHaveLength(baseline);
} finally { reads.mockRestore(); writes.mockRestore(); vi.useRealTimers(); }
const settlePresentation = async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
await notifyChatRunMilestones(tx, { companyId: f.companyId, issueId: f.conversation.issueId! });
await tx.update(heartbeatRuns).set({ nativePhase: "committed", updatedAt: new Date(), resultJson: {
finalizationPhase: "committed", finalizationReasonCode: outcome === "passive" ? "external_chat_response_waiting" : "completed",
nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: outcome === "passive" ? "yielded" : "completed",
continuation: outcome === "passive" ? { kind: "response_wake", idempotencyKey: "settled-presentation" } : null },
presentationDecision: { schema: "paperclip.run_presentation_decision.v1", chosenSource: "none", commentAction: "none",
reasonCodes: [outcome === "passive" ? "yielded_control_plane_wait" : "no_final_response"] },
} }).where(eq(heartbeatRuns.id, f.run.id));
};
await expect(db.transaction(async tx => { await settlePresentation(tx); throw new Error("rollback presentation"); }))
.rejects.toThrow("rollback presentation");
expect(f.responseWrites).toHaveLength(baseline);
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id)))[0].nativePhase).toBe("observed");
await db.transaction(settlePresentation);
await expect.poll(async () => (await readChatActionsById(action.id))[0], { timeout: 5000 }).toMatchObject({
status: outcome === "passive" ? "cancelled" : "processed", result: { final: true },
});
if (outcome === "passive") {
expect(f.responseWrites).toHaveLength(baseline);
expect([...f.responseComments.values()][0].body).toBe(originalBody);
} else expect([...f.responseComments.values()][0].body).toContain("without publishing a final response");
await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), { timeout: 5000 })
.toEqual({ next: null, active: 0 });
expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it("wakes terminal GitHub responses after the outer issue transaction commits", async () => {
const f = await toolReplyFixture();
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
const worker = githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true);
try {
await worker.ready;
const baseline = f.responseWrites.length;
const change = async (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => {
await tx.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
await issueService(tx as unknown as Db).update(f.conversation.issueId!, { status: "blocked" });
};
await expect(db.transaction(async tx => { await change(tx); throw new Error("rollback terminal issue"); }))
.rejects.toThrow("rollback terminal issue");
expect(f.responseWrites).toHaveLength(baseline);
let release!: () => void, entered!: () => void;
const hold = new Promise<void>(resolve => { release = resolve; });
const ready = new Promise<void>(resolve => { entered = resolve; });
const pending = db.transaction(async tx => { await change(tx); entered(); await hold; });
try { await ready; expect(f.responseWrites).toHaveLength(baseline); }
finally { release(); await pending; }
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(errors).toEqual([]);
} finally { await coordinator.stop(); }
});
it("keeps an acknowledged paused GitHub request quiet until its new run finishes", async () => {
const f = await toolReplyFixture("github:paperclipai/paperclip:418");
await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id));
await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.conversation.issueId!));
const thread = makeThread({ channelId: "paperclipai/paperclip", id: f.conversation.externalThreadId }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41802", text: "New request", userId: "42", userName: "octocat", mentioned: true }) });
const responses = githubResponseCommentService(db, f.providerFetch);
await responses.processPending();
expect([...f.responseComments.values()][0].body).toContain("paused");
const errors: unknown[] = [];
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
try {
await responses.registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
const providerWrites = f.responseWrites.length;
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
expect(f.responseWrites).toHaveLength(providerWrites);
reads.mockRestore(); writes.mockRestore(); vi.useRealTimers();
await coordinator.stop();
await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.conversation.issueId!));
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.assignedAgentId,
status: "failed", runtimeMode: "native", contextSnapshot: await chatWakeContext({
endpointId: f.endpoint.id, issueId: f.conversation.issueId, provider: "github", providerMessageId: "41802",
}) });
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
await responses.registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull(); expect(errors).toEqual([]);
} finally { await coordinator.stop(); vi.useRealTimers(); }
});
it.each(["restart", "drain", "standby"] as const)("retains terminal GitHub response work across %s", async mode => {
const f = await toolReplyFixture();
let canRun = true;
const errors: unknown[] = [];
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
canReconcile: () => mode !== "standby" || canRun, onError: error => errors.push(error) });
const responses = githubResponseCommentService(db, f.providerFetch);
let worker = responses.registerTerminalResponseWork(coordinator, () => canRun);
try {
await worker.ready;
const baseline = f.responseWrites.length;
if (mode === "restart") await coordinator.stop(); else canRun = false;
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
await issueService(db).update(f.conversation.issueId!, { status: "blocked" });
expect(f.responseWrites).toHaveLength(baseline);
if (mode === "restart") {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
worker = responses.registerTerminalResponseWork(coordinator, () => true);
await worker.ready;
} else { canRun = true; worker.wake(); }
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(errors).toEqual([]);
} finally { await coordinator.stop(); }
});
it("reactivates terminal GitHub response recovery from committed native identity", async () => {
const f = await toolReplyFixture();
const messageId = String(f.run.contextSnapshot?.wakeCommentId);
await db.update(heartbeatRuns).set({ status: "failed", nativeIssueId: f.conversation.issueId,
contextSnapshot: {}, updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
try {
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
const baseline = f.responseWrites.length;
await expect(db.transaction(async tx => {
await initializeRunIdentity(tx as unknown as Db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!,
responsibleUserId: "owner-user", messageIds: [messageId], cause: "instruction" });
throw new Error("rollback identity");
})).rejects.toThrow("rollback identity");
expect(f.responseWrites).toHaveLength(baseline);
await initializeRunIdentity(db, { companyId: f.companyId, runId: f.run.id, issueId: f.conversation.issueId!,
responsibleUserId: "owner-user", messageIds: [messageId], cause: "instruction" });
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(errors).toEqual([]);
} finally { await coordinator.stop(); }
});
it("waits for the existing GitHub publication lease before a terminal response", async () => {
const f = await toolReplyFixture();
const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id));
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
let release!: () => void, entered!: () => void;
const hold = new Promise<void>(resolve => { release = resolve; });
const ready = new Promise<void>(resolve => { entered = resolve; });
let lease: Promise<unknown> | undefined;
try {
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
lease = withGitHubPublicationLease(db, { companyId: f.companyId, endpointId: f.endpoint.id,
repositoryId: String(resource.metadata.providerRepositoryId), number: 418 }, f.providerFetch, async () => { entered(); await hold; });
await ready;
const baseline = f.responseWrites.length;
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
await issueService(db).update(f.conversation.issueId!, { status: "blocked" });
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).not.toBeNull();
expect(f.responseWrites).toHaveLength(baseline);
release(); await lease;
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(errors).toEqual([]);
} finally { release(); await lease; await coordinator.stop(); }
});
it("finds owed terminal GitHub work beyond a full page of unrelated request history", async () => {
const f = await toolReplyFixture("github:paperclipai/paperclip:418");
const [original] = await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
const [delivery] = await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, original.deliveryId!));
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
const history = Array.from({ length: 101 }, (_, index) => ({
actionId: `00000000-0000-4000-8000-${String(index + 1).padStart(12, "0")}`,
deliveryId: randomUUID(), commentId: randomUUID(), providerId: `retained-request-${index}`,
}));
await db.insert(issueComments).values(history.map(row => ({ id: row.commentId, companyId: f.companyId,
issueId: f.conversation.issueId!, authorUserId: "owner-user", body: "A separate retained request" })));
await db.insert(chatDeliveries).values(history.map(row => ({ ...delivery, id: row.deliveryId,
providerEventId: row.providerId, deduplicationKey: row.providerId, state: "processed" as const })));
await db.insert(chatMessageLinks).values(history.map(row => ({ companyId: f.companyId, endpointId: f.endpoint.id,
conversationId: f.conversation.id, deliveryId: row.deliveryId, commentId: row.commentId,
providerMessageId: row.providerId, direction: "inbound" })));
await db.insert(chatActions).values(history.map(row => ({ ...original, id: row.actionId, deliveryId: row.deliveryId,
providerActionId: `github_response:${row.deliveryId}`, result: { code: "fast_response_superseded" } })));
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
try {
const baseline = f.responseWrites.length;
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(f.responseWrites).toHaveLength(baseline + 1);
expect(await db.select().from(chatActions).where(inArray(chatActions.id, history.map(row => row.actionId))))
.toEqual(expect.arrayContaining(history.map(row => expect.objectContaining({ id: row.actionId,
status: "processed", result: { code: "fast_response_superseded" } }))));
expect(errors).toEqual([]);
} finally {
await coordinator.stop();
await db.delete(chatActions).where(inArray(chatActions.id, history.map(row => row.actionId)));
}
});
it.each([false, true])("parks terminal GitHub responses behind the publication owner (uncertain: %s)", async uncertain => {
const f = await toolReplyFixture();
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
let release!: () => void, entered!: () => void;
const hold = new Promise<void>(resolve => { release = resolve; });
const ready = new Promise<void>(resolve => { entered = resolve; });
let intercepted = false;
let sending: Promise<unknown> | undefined;
const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_review_publication"))))[0];
try {
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
f.setSupplementalProviderFetch(async input => {
if (intercepted || !String(input).endsWith("/access_tokens")) return undefined;
const publication = await action();
if (!publication) return undefined;
intercepted = true;
if (uncertain) await db.update(chatActions).set({ result: { replyWriteStarted: true } }).where(eq(chatActions.id, publication.id));
await db.update(heartbeatRuns).set({ status: "failed", updatedAt: new Date() }).where(eq(heartbeatRuns.id, f.run.id));
await issueService(db).update(f.conversation.issueId!, { status: "blocked" });
entered(); await hold;
return undefined;
});
const baseline = f.responseWrites.length;
sending = githubChatReviewService(db, f.providerFetch).execute(f.session, "comment", { body: "Authored final response", idempotencyKey: "terminal-owner-race" });
await ready;
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(f.responseWrites).toHaveLength(baseline);
release(); await sending;
expect((await action()).status).toBe("cancelled");
if (!uncertain) await expect.poll(() => [...f.responseComments.values()][0]?.body, { timeout: 5000 }).toContain("stopped before completing");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
if (uncertain) expect(f.responseWrites).toHaveLength(baseline);
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
await vi.advanceTimersByTimeAsync(120_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
reads.mockRestore(); writes.mockRestore(); expect(errors).toEqual([]);
} finally { release(); await sending; await coordinator.stop(); vi.useRealTimers(); }
});
it.each(["tree restore", "workspace admission", "exhausted recovery"] as const)("wakes a paused GitHub response from %s", async producer => {
const f = await toolReplyFixture("github:paperclipai/paperclip:418");
const issueId = f.conversation.issueId!;
await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, f.run.id));
await db.update(chatActions).set({ result: sql`coalesce(${chatActions.result}, '{}'::jsonb) || '{"final":true}'::jsonb` })
.where(and(eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "github_response_comment")));
const thread = makeThread({ channelId: "paperclipai/paperclip", id: f.conversation.externalThreadId }).thread;
await deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id, provider: "github", thread,
trigger: "mention", message: makeMessage({ id: "41802", text: "New request", userId: "42", userName: "octocat", mentioned: true }) });
const [link] = await db.select().from(chatMessageLinks).where(and(eq(chatMessageLinks.endpointId, f.endpoint.id), eq(chatMessageLinks.providerMessageId, "41802")));
const action = async () => (await db.select().from(chatActions).where(and(eq(chatActions.deliveryId, link.deliveryId!), eq(chatActions.kind, "github_response_comment"))))[0];
let mutate: () => Promise<unknown>;
if (producer === "tree restore") {
await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, issueId));
const tree = issueTreeControlService(db);
const actor = { actorType: "user" as const, actorId: "owner-user", userId: "owner-user" };
const cancelled = await tree.createHold(f.companyId, issueId, { mode: "cancel", actor });
await tree.cancelIssueStatusesForHold(f.companyId, issueId, cancelled.hold.id);
const restore = await tree.createHold(f.companyId, issueId, { mode: "restore", actor });
mutate = () => tree.restoreIssueStatusesForHold(f.companyId, issueId, restore.hold.id, { actor });
} else if (producer === "workspace admission") {
const settings = instanceSettingsService(db);
const previous = await settings.getExperimental();
await settings.updateExperimental({ enableIsolatedWorkspaces: true });
onTestFinished(() => settings.updateExperimental({ enableIsolatedWorkspaces: previous.enableIsolatedWorkspaces }));
const heartbeat = heartbeatService(db);
onTestFinished(() => heartbeat.drainActiveRunExecutions());
await db.update(issues).set({ status: "todo", workMode: "standard", responsibleUserId: "owner-user",
projectId: null, projectWorkspaceId: null, executionWorkspaceId: null, executionRunId: null, checkoutRunId: null,
executionWorkspaceSettings: { mode: "isolated_workspace", workspaceStrategy: { type: "git_worktree" } },
}).where(eq(issues.id, issueId));
await db.update(agents).set({ runtimeConfig: { heartbeat: { wakeOnDemand: true } } }).where(eq(agents.id, f.assignedAgentId));
mutate = () => heartbeat.wakeup(f.assignedAgentId, { source: "assignment", triggerDetail: "system",
reason: "issue_assigned", payload: { issueId }, contextSnapshot: { issueId, wakeReason: "issue_assigned" } });
} else {
await db.update(issues).set({ status: "in_progress", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId));
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: issueId,
kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: f.assignedAgentId,
cause: "uncertain_external_action", status: "active", evidence: { runId: f.run.id }, fingerprint: randomUUID(),
nextAction: "Preserve the recorded outcome without replay.",
});
mutate = () => settleUnrecoverableExecutions(db);
}
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
try {
await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready;
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect((await action()).result?.pausedVersion).toBeUndefined();
await mutate();
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].status).toBe("blocked");
await expect.poll(async () => (await action()).result?.pausedVersion, { timeout: 5000 }).toEqual(expect.any(String));
expect([...f.responseComments.values()][0].body).toContain("paused");
await expect.poll(() => coordinator.nextWakeAt(), { timeout: 5000 }).toBeNull();
expect(errors).toEqual([]);
} finally { await coordinator.stop(); }
});
it.each(["release", "restart", "drain", "standby"] as const)("wakes GitHub acknowledgements from a real publication lease %s", async mode => {
const f = await reviewBotFixture();
const [resource] = await db.select().from(chatEndpointResources).where(eq(chatEndpointResources.endpointId, f.endpoint.id));
+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.registerGitHubTerminalResponseWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerGitHubAcknowledgementWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerTerminalConfirmationWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
+1
View File
@@ -3504,6 +3504,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
return {
registerInteractionContinuationWork,
registerGitHubTerminalResponseWork: githubResponseCommentService(db, fetchImpl).registerTerminalResponseWork,
registerGitHubAcknowledgementWork: githubResponseCommentService(db, fetchImpl).registerAcknowledgementWork,
registerTerminalConfirmationWork,
loadFailedRetryNoticeScopes,
@@ -1,5 +1,4 @@
import type { Db } from "@paperclipai/db";
import { githubResponseCommentService } from "../../chat-github-response-comments.js";
import { githubReviewCheckService } from "../../chat-github-checks.js";
export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fetch, options: {
@@ -13,7 +12,6 @@ export function createGitHubMaintenance(db: Db, fetchImpl: typeof globalThis.fet
options.scheduleMessageProcessing(async () => {
try {
// Provider I/O must not block Slack, Teams, or ordinary message receipts.
await githubResponseCommentService(db, fetchImpl).processTerminalResponses();
await githubReviewCheckService(db, fetchImpl).processPending();
} finally {
githubMaintenancePending = false;
@@ -0,0 +1,42 @@
import type { Db } from "@paperclipai/db";
import { githubRunReplyState } from "../../chat-run-publications.js";
import { readGitHubResponseRun } from "./response-run-state.js";
export { responseRunColumns as githubResponseRunColumns } from "./response-run-state.js";
export type GitHubResponseExecutionInput = {
companyId: string;
endpointId: string;
agentId: string;
issueId: string;
deliveryId: string;
sourceMessageId: string | null;
issueStatus: string;
deliveryCreatedAt: Date;
};
export async function githubResponseExecution(db: Db, input: GitHubResponseExecutionInput) {
const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, input);
const requestRunning = ownsLatestRequest && latestRun?.status === "running";
const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status);
const run = terminal && (ownsLatestRequest || input.issueStatus === "blocked") ? latestRun : null;
const reply = run && ownsLatestRequest ? await githubRunReplyState(db, {
companyId: input.companyId, endpointId: input.endpointId, issueId: input.issueId, runId: run.id,
}) : "none";
const paused = !!run ? input.issueStatus === "blocked" && !ownsLatestRequest
: input.issueStatus === "blocked" && latestRun?.status !== "running";
const presentationPending = !!run && ownsLatestRequest && run.nativePresentationPending && reply === "none";
const passiveResponseWait = !!run && ownsLatestRequest && run.passiveResponseWait && reply === "none";
const body = presentationPending ? undefined : run ? reply === "confirmed" ? "This request has been handled. See the response in this conversation."
: paused ? "Work is paused. Your request is saved. See the Paperclip task for details."
: run.status === "succeeded" ? "This turn ended without publishing a final response. See the Paperclip task for details."
: "This turn stopped before completing. See the Paperclip task for details."
: paused ? "Work is paused. Your request is saved. See the Paperclip task for details." : undefined;
return {
body, paused, reply, running: requestRunning, presentationPending, passiveResponseWait,
versionAt: run ? new Date(Math.max(run.updatedAt.getTime(), input.deliveryCreatedAt.getTime()))
: paused ? input.deliveryCreatedAt : undefined,
version: JSON.stringify([latestRun?.id ?? null, latestRun?.status ?? null,
latestRun?.updatedAt.toISOString() ?? null, ownsLatestRequest, input.issueStatus === "blocked", reply]),
};
}
@@ -0,0 +1,106 @@
import { and, asc, eq, gt, isNotNull, sql } from "drizzle-orm";
import { chatActions, chatConversations, chatDeliveries, chatEndpointLeases, chatEndpointResources,
chatEndpoints, chatMessageLinks, heartbeatRuns, issues, type Db } from "@paperclipai/db";
import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js";
import type { ChatResponseScope } from "../../chat-response-availability.js";
import { registerScopedDeadlineWork } from "../../scoped-deadline-work.js";
import { githubResponseExecution } from "./response-execution.js";
import { HttpError } from "../../../errors.js";
type Scope = ChatResponseScope & { endpointId?: string; repositoryId?: string; number?: number };
type Entry = { id: string; companyId: string; endpointId: string; deliveryId: string;
issueId: string; conversationId: string; repositoryId: string | null; number: number | null; dueAt: number };
export function registerGitHubTerminalResponseWork(
db: Db,
coordinator: ReturnType<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]*)?$')`;
async function page(scope?: Scope, cursor?: string, actionId?: string) {
return db.select({
id: chatActions.id, companyId: chatActions.companyId, endpointId: chatActions.endpointId,
deliveryId: chatActions.deliveryId, conversationId: chatConversations.id, issueId: issues.id,
agentId: chatEndpoints.assignedAgentId, issueStatus: issues.status,
deliveryCreatedAt: chatDeliveries.createdAt, repositoryId, number, result: chatActions.result,
sourceMessageId: sql<string | null>`(select l.comment_id from ${chatMessageLinks} l
where l.company_id = ${chatActions.companyId} and l.endpoint_id = ${chatActions.endpointId}
and l.delivery_id = ${chatActions.deliveryId} and l.direction = 'inbound' limit 1)`,
leaseExpiresAt: sql<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})`,
}).from(chatActions)
.innerJoin(chatConversations, and(eq(chatConversations.id, chatActions.conversationId), eq(chatConversations.companyId, chatActions.companyId)))
.innerJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId)))
.innerJoin(chatDeliveries, and(eq(chatDeliveries.id, chatActions.deliveryId), eq(chatDeliveries.companyId, chatActions.companyId)))
.innerJoin(issues, and(eq(issues.id, chatConversations.issueId), eq(issues.companyId, chatActions.companyId)))
.leftJoin(chatEndpointResources, and(eq(chatEndpointResources.id, chatConversations.resourceId), eq(chatEndpointResources.companyId, chatActions.companyId)))
.where(and(eq(chatActions.kind, "github_response_comment"), eq(chatActions.status, "processed"),
isNotNull(chatActions.deliveryId), isNotNull(chatEndpoints.assignedAgentId),
sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`,
sql`(${issues.status} = 'blocked' or (select r.status from ${heartbeatRuns} r
where r.company_id = ${chatActions.companyId} and r.agent_id = ${chatEndpoints.assignedAgentId}
and coalesce(r.native_issue_id::text, r.context_snapshot->>'issueId') = ${chatConversations.issueId}::text
order by r.created_at desc, r.id desc limit 1) in ('succeeded', 'failed', 'interrupted', 'timed_out', 'cancelled'))`,
scope ? eq(chatActions.companyId, scope.companyId) : undefined,
scope?.issueId ? eq(issues.id, scope.issueId) : undefined,
scope?.conversationId ? eq(chatActions.conversationId, scope.conversationId) : undefined,
scope?.deliveryId ? eq(chatActions.deliveryId, scope.deliveryId) : undefined,
scope?.endpointId ? eq(chatActions.endpointId, scope.endpointId) : undefined,
scope?.repositoryId ? eq(repositoryId, scope.repositoryId) : undefined,
scope?.number ? eq(number, String(scope.number)) : undefined,
cursor ? gt(chatActions.id, cursor) : undefined, actionId ? eq(chatActions.id, actionId) : undefined,
)).orderBy(asc(chatActions.id)).limit(100);
}
async function entry(row: Awaited<ReturnType<typeof page>>[number]): Promise<Entry | null> {
const pending: Entry = { id: row.id, companyId: row.companyId, endpointId: row.endpointId, deliveryId: row.deliveryId!,
issueId: row.issueId, conversationId: row.conversationId, repositoryId: row.repositoryId,
number: row.number ? Number(row.number) : null, dueAt: row.leaseExpiresAt ? Date.parse(row.leaseExpiresAt) : 0 };
try {
const execution = await githubResponseExecution(db, { ...row, agentId: row.agentId!, deliveryId: row.deliveryId! });
if (execution.body === undefined || execution.reply === "unsettled" ||
(execution.paused && row.result?.pausedVersion === execution.version)) return null;
} catch (error) {
if (!(error instanceof HttpError) || error.status !== 403) throw error;
// Keep this request eligible for the acknowledgement owner's authority
// check and cancellation; it must not poison discovery for other requests.
}
return pending;
}
return registerScopedDeadlineWork(coordinator, {
queue: DELIVERY_QUEUES.chatGitHubTerminalResponses,
decodeScope: (hint: string): Scope | undefined => hint === "*" ? undefined : JSON.parse(hint),
matches: (value: Entry, scope?: Scope) => !scope || (value.companyId === scope.companyId &&
(!scope.issueId || value.issueId === scope.issueId) && (!scope.deliveryId || value.deliveryId === scope.deliveryId) &&
(!scope.conversationId || value.conversationId === scope.conversationId) &&
(!scope.endpointId || value.endpointId === scope.endpointId) && (!scope.repositoryId || value.repositoryId === scope.repositoryId) &&
(!scope.number || value.number === scope.number)),
load: async (scope?: Scope, signal?: AbortSignal) => {
const entries: Entry[] = [];
let cursor: string | undefined;
while (canRun() && !signal?.aborted) {
const rows = await page(scope, cursor);
for (const row of rows) {
if (!canRun() || signal?.aborted) throw signal?.reason ?? new Error("GitHub terminal-response recovery stopped");
const value = await entry(row); if (value) entries.push(value);
}
if (rows.length < 100) return entries;
cursor = rows[rows.length - 1].id;
}
throw signal?.reason ?? new Error("GitHub terminal-response recovery stopped");
},
check: async (value: Entry, signal?: AbortSignal) => {
const [row] = await page(undefined, undefined, value.id);
const current = row ? await entry(row) : null;
if (!current || current.dueAt > Date.now() || !canRun() || signal?.aborted) return current;
await acknowledge(current.deliveryId, () => canRun() && !signal?.aborted);
const [remaining] = await page(undefined, undefined, value.id);
return remaining ? entry(remaining) : null;
},
}, canRun);
}
@@ -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.chatGitHubTerminalResponses, JSON.stringify(scope));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify(scope));
await tx.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.id, claimed.id), eq(chatEndpointLeases.token, token)));
});
@@ -1,4 +1,3 @@
import { readGitHubResponseRun, responseRunColumns } from "./chat-channels/github/response-run-state.js";
import { and, asc, desc, eq, inArray, or, sql } from "drizzle-orm";
import {
chatActions, chatConversations, chatDeliveries, chatEndpoints,
@@ -6,6 +5,8 @@ import {
fastResponseRequests, chatPublications,
type Db,
} from "@paperclipai/db";
import { registerGitHubTerminalResponseWork } from "./chat-channels/github/terminal-response-work.js";
import { githubResponseExecution, githubResponseRunColumns as responseRunColumns } from "./chat-channels/github/response-execution.js";
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { registerGitHubAcknowledgementWork } from "./chat-channels/github/acknowledgement-work.js";
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
@@ -78,7 +79,7 @@ async function stageResponse(db: Db, source: Source) {
export async function writeGitHubResponseComment(
db: Db, source: Source, api: Api, lease: Lease,
input: { body: string; final: boolean; versionAt?: Date; acknowledgement?: boolean; preserveFinal?: boolean },
input: { body: string; final: boolean; versionAt?: Date; acknowledgement?: boolean; preserveFinal?: boolean; pausedVersion?: string },
assertCurrent: () => Promise<void>,
) {
await assertCurrent();
@@ -101,8 +102,10 @@ export async function writeGitHubResponseComment(
(input.versionAt && Date.parse(String(response.result.versionAt)) > input.versionAt.getTime()))) {
if (!input.acknowledgement && response.result.final === true && !input.final)
throw conflict("This GitHub response is already final");
if (["received", "failed"].includes(response.status)) await lease.commit(async tx => {
await tx.update(chatActions).set({ status: "processed", updatedAt: new Date() })
if (["received", "failed"].includes(response.status) || (response.status === "processed" &&
input.pausedVersion && response.result.pausedVersion !== input.pausedVersion)) await lease.commit(async tx => {
await tx.update(chatActions).set({ status: "processed",
result: { ...response.result, ...(input.pausedVersion ? { pausedVersion: input.pausedVersion } : {}) }, updatedAt: new Date() })
.where(and(eq(chatActions.id, response.id), eq(chatActions.status, response.status)));
});
return { id: String(response.result.id), url: String(response.result.url) };
@@ -147,6 +150,7 @@ export async function writeGitHubResponseComment(
}, updatedAt: new Date() }).where(eq(chatActions.id, target.id));
await tx.update(chatActions).set({ status: "processed", result: {
...receipt, final: input.final || alreadyFinal, versionAt: versionAt.toISOString(),
...(input.pausedVersion ? { pausedVersion: input.pausedVersion } : {}),
}, updatedAt: new Date() }).where(eq(chatActions.id, response.id));
await logActivity(tx as unknown as Db, { companyId: source.endpoint.companyId,
actorType: "system", actorId: "github-response", action: "chat_github.response_updated",
@@ -207,52 +211,41 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
let requestRunning = false;
let requestPaused = false;
let preserveFinal = false;
let pausedVersion: string | undefined;
if (finalBody === undefined) {
const existing = await stageResponse(db, source);
const { latestRun, ownsLatestRequest } = await readGitHubResponseRun(db, {
const execution = await githubResponseExecution(db, {
companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
agentId: source.endpoint.assignedAgentId, issueId: source.conversation.issueId,
deliveryId, sourceMessageId: source.sourceMessageId,
agentId: source.endpoint.assignedAgentId!, issueId: source.conversation.issueId!, deliveryId,
sourceMessageId: source.sourceMessageId, issueStatus: source.issueStatus,
deliveryCreatedAt: source.delivery.createdAt,
});
requestRunning = ownsLatestRequest && latestRun?.status === "running";
const terminal = latestRun && ["succeeded", "failed", "interrupted", "timed_out", "cancelled"].includes(latestRun.status);
const run = terminal && (ownsLatestRequest || source.issueStatus === "blocked") ? latestRun : null;
if (run) {
const reply = ownsLatestRequest ? await githubRunReplyState(db, {
companyId: source.endpoint.companyId, endpointId: source.endpoint.id,
issueId: source.conversation.issueId!, runId: run.id }) : "none";
if (reply === "unsettled") throw conflict("GitHub reply is still being resolved");
// A confirmed reply for this exact run can retire its placeholder.
// Otherwise terminal execution is not yet a missing-response decision.
if (ownsLatestRequest && run.nativePresentationPending && reply === "none") return;
if (isMainPullResponse(source) && ownsLatestRequest && run.passiveResponseWait && reply === "none") {
// A committed passive turn intentionally has no new response.
// Settle this request's acknowledgement without editing the shared
// PR assessment or inventing a current-run publication receipt.
await lease.commit(async tx => { await tx.update(chatActions).set({
status: "cancelled", result: { ...existing.result, final: true, code: "passive_response_wait" },
updatedAt: new Date(),
if (execution.reply === "unsettled") throw conflict("GitHub reply is still being resolved");
if (execution.presentationPending) return;
if (isMainPullResponse(source) && execution.passiveResponseWait) {
// Preserve the shared assessment when this exact run intentionally waits
// without publishing. Retire only its request acknowledgement.
await lease.commit(async tx => {
await notifyChatResponseAvailability(tx, { companyId: source.endpoint.companyId, deliveryId });
await tx.update(chatActions).set({ status: "cancelled",
result: { ...existing.result, final: true, code: "passive_response_wait" }, updatedAt: new Date(),
}).where(and(eq(chatActions.id, existing.id), eq(chatActions.companyId, source.endpoint.companyId),
sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`)); });
return;
}
if (existing.result?.final === true) return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined;
preserveFinal = reply === "confirmed";
onlyExisting = reply === "confirmed" && !existing.result?.id;
requestPaused = source.issueStatus === "blocked" && !ownsLatestRequest;
finalBody = reply === "confirmed" ? "This request has been handled. See the response in this conversation."
: requestPaused ? "Work is paused. Your request is saved. See the Paperclip task for details."
: run.status === "succeeded" ? "This turn ended without publishing a final response. See the Paperclip task for details."
: "This turn stopped before completing. See the Paperclip task for details.";
finalVersionAt = new Date(Math.max(run.updatedAt.getTime(), source.delivery.createdAt.getTime()));
} else if (source.issueStatus === "blocked" && latestRun?.status !== "running") {
requestPaused = true;
finalBody = "Work is paused. Your request is saved. See the Paperclip task for details.";
finalVersionAt = source.delivery.createdAt;
} else if (existing.result?.id && (!isMainPullResponse(source) ||
sql`coalesce(${chatActions.result}->>'final', 'false') <> 'true'`));
});
return;
}
if (existing.result?.final === true) return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined;
if (execution.paused && existing.status === "processed" && existing.result?.id && existing.result?.pausedVersion === execution.version)
return existing.result.id ? { id: String(existing.result.id), url: String(existing.result.url) } : undefined;
pausedVersion = execution.paused ? execution.version : undefined;
requestRunning = execution.running;
requestPaused = execution.paused;
preserveFinal = execution.reply === "confirmed";
onlyExisting = preserveFinal && !existing.result?.id;
finalBody = execution.body;
finalVersionAt = execution.versionAt;
if (finalBody === undefined && existing.result?.id && (!isMainPullResponse(source) ||
(await stagePullResponse(db, source)).result?.id === existing.result.id)) {
// Rotate no-op maintenance checks so a page of newer queued asks
// cannot hide other conversations that need terminal reconciliation.
await lease.commit(async tx => { await tx.update(chatActions).set({
status: ["received", "failed"].includes(existing.status) ? "processed" : existing.status,
updatedAt: new Date(),
@@ -293,7 +286,7 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
}
return writeGitHubResponseComment(db, source, api, lease,
{ body: finalBody ?? (isMainPullResponse(source) && !requestRunning ? "Review queued…" : "Working on this…"), final: finalBody !== undefined && !requestPaused,
acknowledgement: finalBody === undefined, preserveFinal,
acknowledgement: finalBody === undefined, preserveFinal, pausedVersion,
...(finalBody !== undefined ? { versionAt: finalVersionAt ?? new Date() } : {}) }, assertCurrent);
});
}
@@ -454,6 +447,8 @@ export function githubResponseCommentService(db: Db, fetchImpl = fetch) {
}
}
return { tryAcknowledge, processPending, processTerminalResponses, failRun, finishRun, hasWorkingResponseForRun,
registerTerminalResponseWork: (coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) =>
registerGitHubTerminalResponseWork(db, coordinator, tryAcknowledge, canRun),
registerAcknowledgementWork: (coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) =>
registerGitHubAcknowledgementWork(db, coordinator, tryAcknowledge, canRun),
};
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { githubRunReplyState } from "./chat-run-publications.js";
import { stageReceiptReactionRemovals } from "./chat-receipt-reactions.js";
import { chatRunSourceMessageId } from "./chat-channels/run-origin.js";
@@ -461,6 +462,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
): Promise<string> {
if (writer === db) return db.transaction(tx => stage(source, operation, payload, key, invocationId, toolName, parameters, tx));
const actionKey = `github_publication:${source.issue.id}:${operation}:${hash(key)}`;
await notifyChatResponseAvailability(writer, { companyId: source.endpoint.companyId, issueId: source.issue.id });
await notifyDeliveryWork(writer, DELIVERY_QUEUES.chatGitHubPublications);
const values = {
companyId: source.endpoint.companyId,
@@ -1046,6 +1048,7 @@ export function githubChatReviewService(db: Db, fetchImpl = fetch) {
)?.[1];
if (!number || !binding?.repository.providerRepositoryId) {
await db.transaction(async tx => {
await notifyChatResponseAvailability(tx, { companyId: action.companyId, conversationId: action.conversationId ?? undefined });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubPublications);
await tx.update(chatActions).set({ status: "cancelled", result: { code: "publication_binding_unavailable", retryable: false }, updatedAt: new Date() })
.where(and(eq(chatActions.id, actionId), eq(chatActions.companyId, action.companyId)));
@@ -3,5 +3,6 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string };
export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId }));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId }));
}
@@ -9,6 +9,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio
export type ChatRunMilestoneScope = { companyId: string; issueId?: string };
export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatRunMilestones, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
// Lifecycle and coarse native progress share lineage and authority. Extend
// this eligibility boundary once rather than adding wrappers to its callers.
@@ -41,6 +41,7 @@ export const DELIVERY_QUEUES = {
chatProviderEffects: "chat-provider-effects",
chatGitHubIngress: "chat-github-ingress",
chatGitHubPublications: "chat-github-publications",
chatGitHubTerminalResponses: "chat-github-terminal-responses",
chatGitHubAcknowledgements: "chat-github-acknowledgements",
chatFailedRunRetries: "chat-failed-run-retries",
chatTelegramMaintenance: "chat-telegram-maintenance",
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyExecutionRunWork } from "./execution-control-notifications.js";
import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
@@ -587,6 +588,7 @@ export async function settleUnrecoverableExecutions(
let nativeFailureBlock = action.evidence.nativeFailureBlock;
let nativeBootstrapFailureBlock = action.evidence.nativeBootstrapFailureBlock;
if (current) {
if (task.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: task.companyId, issueId: task.id });
const [projected] = await tx
.update(issues)
.set({
+2
View File
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "../chat-response-availability.js";
import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
import { assertTaskRetryConversationAllowed, readTaskRetryIntent, readTaskRetryBoardMessageOrigin, retireSupersededTaskRetry, taskRetryMutationAllowed } from "../task-retry.js";
@@ -3863,6 +3864,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
`- Reason: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_MESSAGE}`,
`- Next action: ${WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION}`,
].join("\n");
if (issue.status !== "blocked") await notifyChatResponseAvailability(tx, { companyId: issue.companyId, issueId: issue.id });
await tx
.update(issues)
.set({
+5 -1
View File
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@@ -999,7 +1000,10 @@ export function issueTreeControlService(db: Db) {
const restored: TreeStatusUpdateResult["updatedIssues"] = [];
for (const [status, issueIdsForStatus] of issueIdsByStatus) {
if (issueIdsForStatus.length === 0) continue;
for (const issueId of issueIdsForStatus) await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`);
for (const issueId of issueIdsForStatus) {
await notifyTerminalWorkspaceWork(tx, `issue:${issueId}`);
await notifyChatResponseAvailability(tx, { companyId, issueId });
}
const rows = await tx
.update(issues)
.set({
+5
View File
@@ -1,3 +1,4 @@
import { notifyChatResponseAvailability } from "./chat-response-availability.js";
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js";
import { notifyApprovalWork } from "./approval-work-notifications.js";
@@ -11430,6 +11431,10 @@ export function issueService(db: Db) {
if (patch.status === "done" || patch.status === "cancelled") {
await notifyDecisionTargetWork(tx, `issue:${id}`);
}
if ((patch.status !== undefined && patch.status !== receiptExisting.status) ||
(patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId)) {
await notifyChatResponseAvailability(tx, { companyId: receiptExisting.companyId, issueId: id });
}
if (patch.status !== undefined && patch.status !== receiptExisting.status) {
await notifyChatInboundWakeNotices(tx, { companyId: receiptExisting.companyId, issueId: id });
}