diff --git a/doc/idle-discord-command-registration.md b/doc/idle-discord-command-registration.md new file mode 100644 index 00000000000..bbbf7cf1ae0 --- /dev/null +++ b/doc/idle-discord-command-registration.md @@ -0,0 +1,210 @@ +# Discord command registration while idle + +Command registration uses the shared delivery coordinator and existing saved +registration deadlines. It no longer runs as a side effect of the recurring +Gateway runtime reconciliation. Foreground configure and resume still attempt +registration immediately under their existing credential lease. + +## Existing obligations and authority + +The registration store owns `chat_actions` records of kind +`discord_command_registration`. The existing `chat_discord_command_owners` table +retains the application namespace owner even after endpoint or company deletion. +No table, trigger, listener connection, or private timer is added. + +A valid successful receipt has a five-minute remote verification deadline. Other +saved outcomes retain their existing retryAt, including Retry-After. An indefinite +retry remains parked. A genuinely unowned namespace is immediately due. Invalid +receipts, foreign owners, and orphaned ownership tombstones do not become new +registration work. The deadline reader is only a hint; the existing protocol still +checks authority, generation, credentials, identity, and receipt state before +provider operations and commits. An uncertain create is recovered by reading +remote state, never by blindly repeating a create. + +Eligible endpoints are Discord, verifying/active/attention, with an active enabled +connection and a known bot/application ID and guild ID. Attention plus failed +connection health remains excluded. These are the old runtime discovery filters; +company status and unrelated endpoint metadata do not add new policy. + +## Scheduling and lifecycle + +Startup reconstructs eligible state in UUID keyset pages of 100. Committed hints +refresh only the affected endpoint, connection, or deleted company. Refresh and +retry passes have a 30-second minimum spacing when no already-due backlog +remains. A bounded batch with more due rows continues after one second. There is no periodic inventory +scan after all hints settle. Due dispatch rechecks persisted eligibility and the +saved deadline, invokes the command owner, and reloads its durable result. + +Each pass admits at most 25 commands, taking turns across companies and preserving +oldest-first order within each company. A company cursor rotates between passes, +including when more companies are due than fit in one pass. A failed or no-progress +attempt backs off from 30 seconds to five minutes; a future stored deadline wins. +A returned promise alone does not prove registration settled. + +Dirty scopes received during work remain pending. Drain and warm standby retain +requests without passive SQL. Discovery checks cancellation between rows/pages. +Shutdown stops the coordinator before the chat service and joins an already +admitted provider operation through settlement. A local Gateway is refreshed +through its existing owner after command work; command failure does not replace +or tear down a healthy listener. + +## Producer coverage + +| Producer | Wake contract | +| --- | --- | +| Registration store | Register intent before initial owner/action insert, state/receipt transitions, and retry-result updates in their existing transaction | +| Configure, reconnect, resume, pause, remove | Existing endpoint-work helper now also signals command scheduling by connection ID | +| Setup test completion | The same endpoint-work helper covers activation | +| Credential reference publication/removal | Signal before binding and reference writes, while retaining the credential lease | +| Gateway connecting, ready, availability and failure callbacks | Scoped endpoint signal when health/status can change eligibility; existing runtime generation fences remain | +| Provider availability reconciliation | Signal available/restored and disabled/removed endpoints | +| Provider-effect, publication and interaction failures | Signal before endpoint attention and failed-health writes | +| Accepted inbound task mutation | Signal the endpoint before its status/connection activation writes; message metadata-only writers stay exempt | +| Generic tool connection API | Existing chat connection transaction helper covers enable/disable and credential/configuration changes | +| Generic connection removal | Signal the SDK connection before archival | +| Generic manual health observation | Chat-owned dispatcher signals SDK connections; other transports do not wake this worker | +| Membership/credential revocation | SDK connections signal before reference removal or disablement | +| Company deletion | Signal the company before the company cascade; namespace tombstones survive | + +Draft endpoint creation has no eligible identity and waits for configuration. +Endpoint display names, reach preferences, resource metadata, publication/event +timestamps, and capability-only runtime projection do not create a registration +obligation. Other providers' setup, OAuth, and registration protocols do not +become Discord work. Direct SQL and independent writers still require explicit +wake integration; process-local hints are not cross-replica notification delivery. + +## Verification and limits + +Selected local verification covers 40 real-PostgreSQL protocol/lifecycle cases, +three fair-batch cases, 93 Discord integration cases, 487 tool-access/access +consumers, and 20 chat-worker/startup consumers: 643 unique tests. Workspace +typechecking and build pass; the final small fairness addition also receives +server type/build verification before publication. The full local `pnpm test:run` +is not claimed: embedded PostgreSQL is unavailable on this host. Remote CI must +complete the full matrix before acceptance. + +The large-backlog fairness regression failed before selection was repaired. +Other observed failures were fixture issues: duplicate live-bot IDs, uncertain +transactions retained by prior fault tests, installing a fake clock after creating +a real timer, pausing a connection still in setup, and removing a fixture twice. +Repeated broad runs on a reused database also found an unchanged hard-coded bot +ID and a global comment-body assertion reading earlier fixtures. The final broad +Discord run uses a fresh schema-only disposable database and passes all 91 cases; +production constraints and the unrelated assertions were not relaxed. See the +[dated audit](plans/2026-10-25-discord-command-registration-sleep-audit.md). + +This migration proves the command worker's quiet periods. Runtime inventory, +receipt-to-runtime projection, active Gateway lease renewal, persistent sockets, +voice, Teams, and other chat maintenance remain separate obligations. The remaining +runtime poll can still read command receipts; it no longer initiates command +registration or remote verification. Cloud sleep eligibility and durable alarm +coverage are unchanged. A timer in this process cannot wake a stopped host. + + +Review follow-up: a 300-command regression failed under the old 25-per-30-second +limit. Bounded one-second continuations now drain the controlled due backlog +within 15 seconds, then remain quiet until real verification deadlines. An unknown +scope hint also reproduced loss of a later valid wake. Parsing now occurs inside +the recovery boundary; an invalid scope requests full reconstruction rather than +dropping queued scopes or retrying a poison hint forever. Both regressions and +all 62 combined command/health/batching cases pass. + +The real producer integration covers credential rotation and company deletion in +addition to configure/removal. The deletion test first lacked a configured agent +lifecycle worker; after adding the controlled host driver it exposed an existing +restrictive endpoint-to-agent foreign key failure. Prerequisite #17561 fixes the +deletion order in its own PR. The command intent is now registered before endpoint +cleanup in the owning transaction. The test verifies deadline cancellation, SQL +silence, and retained ownership tombstones. All seven automatic-registration +integration cases pass; 1,413 deselected cases are not counted as executed. + +A test invocation during prerequisite merging failed before execution because an +import conflict was unresolved. The conflict was repaired preserving both imports, +then the test ran successfully. Final server typecheck/build pass. Fresh CI and +reviews are still required after publication; prior-head Greptile 5/5 and Actions +results do not establish acceptance of this head. + + +The prerequisite's history review moved cleanup ahead of issue deletion as well. +The dependent signal moves with it, before the first endpoint delete. All seven +producer integrations and six expanded deletion cases pass after propagation; +rollback restores conversation/publication history. The finding remains recorded +in the prerequisite document rather than treating the earlier endpoint-only +fixture as sufficient evidence. + + +## Failure-path capacity and CI isolation repair + +Required CI at 58a1ac7595 found one command-inventory select after the removal test +had observed `nextWakeAt() === null`. That scheduler observation can occur during +an admitted reload; it does not prove notification settlement. The assertion then +threw before restoring DB spies and contaminated a later Slack grant-transaction +test with recursive spies. The quiet-window tests now wait for both no tracked +active work and no deadline, and restore spies in `finally`. The downstream Slack +test is included in verification. The CI failure, rather than a local reproducer, +is the evidence for the original race; no timeout or SQL-quiet assertion was relaxed. + +Review also identified a genuine failed-item throughput bug: the shared +coordinator ignored `nextRunAt` after a thrown sweep and always waited its fallback +retry. A controlled 300-command regression with one persistent credential failure +reproduced only 25 attempted commands within 15 seconds. The coordinator now keeps +its fallback scheduled, then allows an explicitly future worker deadline to +preempt it while dispatch is admitted. An overdue/null/unreadable deadline cannot +cause a tight error retry loop; drain retains the ordinary fallback. A failed +deadline reader is reported together with the original failure. Workers without +a deadline keep their existing behavior. Domain retry/backoff remains owned by +the worker; no provider policy enters the coordinator. + +After repair, all 300 commands are attempted within 15 seconds, the failed entry +is attempted once in that window, and its later retry does not recheck the 299 +healthy entries. Five coordinator cases verify future deadlines and fallback +behavior for overdue, later, unreadable, and drain states. All 87 command, +coordinator, and health cases pass; eight real automatic-registration/Slack cases +pass (1,412 other integration cases deselected); 57 delivery/warm-standby/startup +cases pass. Server typecheck/build pass. The health parent is propagated through +company cleanup, retaining both notification paths in the three overlapping +producer sites. The six prerequisite cleanup cases pass. Remote acceptance of +the new Discord head remains required. + +## Updated parent acceptance, October 27 + +A normal merge at `5b50224785252fe49a4501d9695063cf9621e528` includes +company-cleanup parent #17561 at `21be75d2b0` and health #17554 at `8c1d293d1a`. +Only the original 26 Discord-owned paths differ from that parent. The shared +producer sites preserve both health and command notifications; no provider +behavior, schema, or coordinator contract changes in this integration. + +The four complete command-store, company-batching, coordinator, and startup +suites pass 119 cases. Eight real Discord automatic-registration and Slack +held-transaction cases pass separately; 1,581 other chat integration cases are +not counted as executed. These 127 distinct targeted cases use separate isolated +PostgreSQL 17 databases cloned from the migrated health test schema, not an +unmigrated-database fixture proof. The first name filter selected only the Slack +case; the corrected verified group selector then executed all eight cases. +Server typecheck and build pass. Startup tests still emit the recorded +SIGINT/SIGTERM listener-count warnings without assertion failures. + +The earlier failed baselines and remote CI history above remain evidence for +their stated revisions. Fresh current-head reviews and all CI are required, and +both prerequisites must land before this child can merge into master. This does +not establish host suspension, external Cloud alarms, or provider cost savings. + + +## Acceptance after the company-deletion parent landed, October 27 + +The branch integrates master at `f8563d25077b17681ab6ef5949f7b07415a58ede`, +including the landed health and company-deletion repairs. Squash ancestry caused +merge conflicts. Resolution applies only this slice's exact 26-file patch from +its recorded parent `21be75d2b0` to that pinned master. Every other file matches +master. The migration 0394 and current lockfile remain unchanged by this slice. + +All 141 distinct targeted cases pass: 78 command-store, batching, coordinator and +startup-timing cases; 8 real automatic Discord and held-transaction Slack cases; +and 55 complete startup/export and company-deletion cases. The company-deletion +fixture starts on an initially empty isolated database and applies migrations. +The two Discord fixtures use separate migrated database clones. The remaining +1,581 chat integration cases were deselected. Server typecheck and build pass. +The startup suite retains the existing signal-listener warnings without an +assertion failure. Source files did not change after these results; subsequent +edits record this acceptance. No full local matrix or deployed sleep is claimed. +Fresh exact-head reviews and CI remain required before merge. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 250d61f86e1..1e1d8913f72 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -17,13 +17,13 @@ Do not mark an item done from a timer removal or a green check alone. | 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Approval retention [#17497](https://github.com/paperclipai/paperclip/pull/17497) and archive delivery [#17475](https://github.com/paperclipai/paperclip/pull/17475) merged; other source retention pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Continuations [#17478](https://github.com/paperclipai/paperclip/pull/17478) merged; expiry [#17479](https://github.com/paperclipai/paperclip/pull/17479) merged; effects [#17482](https://github.com/paperclipai/paperclip/pull/17482) merged; targets [#17486](https://github.com/paperclipai/paperclip/pull/17486) merged | -| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | +| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | #17554 repaired at `8c1d293d1a`; fresh CI/review in progress | | 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Slices 02–07 merged: [secret proposals](https://github.com/paperclipai/paperclip/pull/17446), [credentials](https://github.com/paperclipai/paperclip/pull/17450), [review expiry](https://github.com/paperclipai/paperclip/pull/17453), [approved recovery](https://github.com/paperclipai/paperclip/pull/17454), [interrupted claims](https://github.com/paperclipai/paperclip/pull/17458), [projection](https://github.com/paperclipai/paperclip/pull/17468) | | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | | 11 | Status-card evaluation | Settings/card mutations and evaluation deadlines | Pending | -| 12 | Plugin scheduled jobs | Registration/update/removal and actual job deadlines | Pending | +| 12 | Plugin scheduled jobs | Registration/update/removal and actual job deadlines | #17518 published; see latest-head gate records below | | 13 | Routine, agent-timer and issue-monitor scheduling | Remove periodic wake backstops after complete producer coverage; routine execution-status reconciliation separately | Pending | | 14 | Skill-version retention and native-harness backup expiration | Separate retention owners and existing retention guarantees | Pending | | 15 | Automatic database backups | Shared scheduling, actual backup deadlines, existing checkpoint contract | Pending | @@ -1314,6 +1314,53 @@ recorded, not counted as passes. Total selected coverage is 770 cases. The serve build passes. The inventory SQL uses the existing company/enabled index with 40,000 retained rows (13 buffers, 0.022 ms). Full workspace and remote gates remain. +## Health publication and next chat boundary + +Tool health is [#17554](https://github.com/paperclipai/paperclip/pull/17554), head +`2ca12f806399cfcef1a474453301534f4b96c1ef`, stacked on #17518. Local workspace +checks/build and 770 selected tests pass. CI run 40165710347 is confirmed live; +Greptile and security review are running. This is not final acceptance. + +The next runtime-discovery audit found command-registration recovery hidden +inside `reconcileDiscordRuntime`. That is a separate durable reason and must get +its own PR before removing the runtime discovery backstop. Slice 19 audits that +owner in [Discord command registration](2026-10-25-discord-command-registration-sleep-audit.md). +Runtime discovery, active lease renewal, voice, Teams, and every original pending +inventory item remain in scope. No merges are authorized. + + +## Slice 19 foundation and health review status + +Discord registration now has a normalized saved-deadline reader, scoped store +notifications, paged eligibility discovery, and a shared-coordinator worker. +Thirty-six real-DB protocol/lifecycle cases and server typechecking pass. See +[the audit and remaining integration](2026-10-25-discord-command-registration-sleep-audit.md). +It is not wired into startup, the old runtime call is not removed, and no PR is +published for this slice yet. Complete producer wiring and consumer verification +before removing the backstop. + +Tool-health #17554 at `da4089960619d3da2112a4eed3cc7a3d97dbebf8` has passing +Actions checks and fresh Greptile 5/5 after the failed-probe warning repair. +Paperclip Review remains queued, so final acceptance is still open. Its PR body +now records 771 selected local tests and the outstanding gate. No PR was merged. + + +## Slice 19 integration + +Discord registration now has complete producer wiring and a startup worker; +recurring runtime reconciliation no longer invokes its command protocol. The +[contract](../idle-discord-command-registration.md) records retained verification +policy, scoped producer coverage, supported writers and separate runtime duties. +Local selected coverage is 639 unique passing tests, with workspace types/build +passing before the final fair-batch addition. Final server typechecking and build +also pass. Remote publication/gates remain open. + +The latest health Paperclip Review completed neutral, with a 4/5 assessment and +follow-ups for cross-company backlog fairness and direct service-producer tests. +These remain owning-PR work despite green Actions and Greptile 5/5. The same +fairness defect was reproduced and repaired in the Discord worker. Shared helper +commit `e77979d4fc` is ready to carry into #17554. No PR is newly accepted or merged. + ## Discord publication and health fairness repair @@ -1365,6 +1412,56 @@ requests preserving queued scopes after malformed hints and broader producer coverage. Its initial head has Greptile 5/5 but these findings remain open. +Discord review repairs are in progress. Both new worker regressions failed before +repair: the 300-command backlog did not drain promptly, and an unknown hint lost +later scopes. Bounded backlog continuations and full reconstruction after malformed +scope parsing repair them; all 62 combined command/health/batching cases pass. +Real credential rotation also passes. The company-deletion producer test exposed +an existing dependency-order failure: agents are deleted before chat endpoints +whose assigned-agent foreign key is RESTRICT. Configuring the fixture lifecycle +removed the initial missing-worker failure but reproduced this actual SQL failure. +Keep the failing producer test; repair the deletion dependency in its own stack +PR before publishing the Discord follow-up. Current local changes are not final +acceptance. The original full inventory and remaining CI gates are still open. + + +Company deletion prerequisite is [#17561](https://github.com/paperclipai/paperclip/pull/17561), +head `4d721402d2`, based on #17554. Both new regressions failed first; all six +cleanup-removal tests and server types/build pass after the fix. The chat owner +removes company endpoints before agent purge in the same transaction. Other +companies and global namespace tombstones remain unchanged; a later failure rolls +back the deletion. Existing voice-history restrictions remain in force. Remote +CI/review are required; this is a focused correctness dependency, not a migration +of another inventory duty. + +The prerequisite is merged into the local Discord branch at `2802c361f3`. The +company command signal precedes endpoint deletion. All seven real automatic +registration integration cases now pass, including rotation and deletion, plus +62 combined worker/batching cases and final server types/build. The earlier +missing lifecycle fixture, actual dependency-order failure, and merge-conflict +invocation are recorded in the Discord document. Update #17557's base to #17561 +and publish the review repairs for fresh CI/review. Cumulative selected Discord +coverage is 643 cases. Health #17554 at `f8ad26f0ff` has no failures and one pending +check at the last observation; do not mark it fully accepted yet. + +Next new owner remains chat runtime discovery/ownership, separate from command +registration, voice, and active transport renewals. The full original inventory, +#17490 E2E cancellation failure, final stack/master reconciliation, and all PR +acceptance gates remain required. No stack merge is authorized. + + +Prerequisite #17561 review found that retained chat conversations/publications +also restrict issue deletion. Expanded real-history fixtures reproduced that +failure, and cleanup now runs immediately after locking the company. All six +cleanup cases pass, including restoration of chat history on later failure. The +review thread is resolved at parent head `e7c29e4840`; fresh review/CI still apply. +Discord inherits that parent at `33954efd2e` and registers its company intent +before the earlier cleanup. Seven producer integrations and all six cleanup cases +pass on the merged branch. The 1,413 deselected broad chat cases are not passes. +Health `f8ad26f0ff` now has fresh Greptile 5/5 and no failed checks; Paperclip Review +was still queued at the last inspection. The full goal remains active. + + Health security repair: five real connection API updates in a 205-connection company reproduced 1,025 inventory rows. Individual writers now signal their connection IDs; application writers signal their application IDs. Genuine broad @@ -1375,6 +1472,17 @@ the repair to #17554, propagate through company-deletion #17561 into Discord #17557, and require fresh remote gates. No PR in this stack is merged. The next new owner remains chat runtime discovery; the full remaining ledger is unchanged. + +Published health security repair is 9ef6e4f4c8; fresh security and Greptile checks +pass, with Actions still running at inspection. Company cleanup inherits it at +e8febb30eb; six cleanup tests pass. Discord inherits it at fe1c886a3d, preserving +both command and scoped health intents. Its previous-head CI exposed a flaky +quiet-window/spying fixture and review reproduced failure-path backlog throttling. +The repaired coordinator honors actual future deadlines without zero-delay error +loops or changing domain backoff. All 152 selected command/coordinator/health, +producer/Slack, delivery/standby/startup cases and server types/build pass. Publish +for fresh CI/review. The full inventory, #17490 CI failure, and stack/master +reconciliation remain open; no merge authority is implied. ## Acceptance follow-up after the plugin parent merged (October 27) Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at @@ -1389,3 +1497,11 @@ baselines, real concurrency proof, and fresh-database verification. Older stack status and test counts above are historical. The original remaining inventory is still in scope beyond the 65 published numbered slices; native backups and the remaining aggregate duties have not been migrated by this follow-up. + + +## Discord registration landing compatibility, October 27 + +Company deletion has landed. The Discord command slice retains only its original +26-file delta against the landed parent and current master at `f8563d2507`. All +141 targeted cases, server typecheck and build pass. The [owning audit](2026-10-25-discord-command-registration-sleep-audit.md#acceptance-after-the-company-deletion-parent-landed-october-27) +records exact scope, fixture setup and remaining remote gates. diff --git a/doc/plans/2026-10-25-configuration-worker-drain-repair.md b/doc/plans/2026-10-25-configuration-worker-drain-repair.md index 99ebdb41834..9b2d8c64483 100644 --- a/doc/plans/2026-10-25-configuration-worker-drain-repair.md +++ b/doc/plans/2026-10-25-configuration-worker-drain-repair.md @@ -29,3 +29,17 @@ Discord command registration and receiver discovery require the same causal check at their own parents, then propagation through every dependent stack PR. The full original recurring-work inventory, milestone producer audit, older E2E failure and inbound cleanup/deletion policy finding remain outstanding. + +## Discord command registration — slice 19 / #17557 + +The strengthened outer-rollback/drain case reproduces active=1 after commit in +/tmp/sleep19-drain-before.log. The repair preserves the dirty scope and saved +deadlines, but reports pending dirty work only while dispatch is allowed. The +shared coordinator still holds unresolved transactions independently. + +All 67 registration/store cases pass in /tmp/sleep19-drain-tests.log, including +90 seconds of no SQL/provider calls and an unchanged idle snapshot during drain, +followed by registration on resume without another producer. Production server +typecheck passes in /tmp/sleep19-drain-typecheck.log. These tests use the migrated +paperclip_sleep19_drain_20261025_test database. Remote CI/review must run on the +new head; this is not evidence of provider suspension. diff --git a/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md b/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md new file mode 100644 index 00000000000..eeea2acdf4c --- /dev/null +++ b/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md @@ -0,0 +1,227 @@ +# Discord command-registration scheduling audit + +Parent: tool-health PR #17554, head `da4089960619d3da2112a4eed3cc7a3d97dbebf8`. +The parent has 771 selected passing local tests, passing workspace typecheck and +build, passing Actions checks, and fresh Greptile 5/5. Paperclip Review completed neutral with a 4/5 assessment and follow-ups for +company fairness and real-producer coverage. It is not accepted or merged. This branch owns +one newly separated reason from the existing chat-runtime inventory. + +## Evidence and boundary + +At the parent revision, `app.ts` calls `reconcileProviderRuntimes` from its recurring chat maintenance +lane. That service discovers eligible Discord and Photon endpoints, restores +runtime ownership, and calls `discord/runtime-reconciliation.ts`. The latter +calls `reconcileDiscordCommands` before inspecting registered command state and +runtime capabilities. Thus removing runtime inventory polling without another +owner would also remove command-registration recovery. + +The command protocol and authorization already live in +`chat-channels/discord/commands.ts`. Durable state is in existing `chat_actions` +with kind `discord_command_registration`, owned by +`chat-discord-command-registration-store.ts`. The store already has retryAt and +retryIndefinite outcomes. A registered receipt has an existing five-minute remote verification deadline; +retain it while eliminating intervening discovery scans. An indefinite hold +must not cause periodic discovery simply because an endpoint is configured. Inspect the full +state protocol before deciding which attempted/unknown states may replay. + +This slice should give command registration its own committed-change/deadline +worker and remove command reconciliation from recurring runtime maintenance. +Keep provider protocol, native identity/credential fences, and command receipt +validation in the Discord owner. Do not copy command HTTP behavior into startup +or the generic scheduler. Keep direct setup/foreground calls compatible. + +## Required producer and lifecycle audit + +- Endpoint creation/configure/reconnect/resume, identity/guild changes, credential + rotation, removal, and connection disable/health quarantine. +- All registration-store claims, attempted/unknown/registered commits, recorded + retry outcomes, legacy restoration, and operator force/reset paths. +- Existing registered command receipt publication and runtime capability refresh: + establish a committed completion trigger without continuing a periodic scan. +- Shared receiver-ownership loss and lease contention: identify which operations + require an owned runtime, a credential lease, or only an eligible endpoint. +- Configuration changes arriving during provider work must retain the existing + generation/fingerprint fence and schedule the latest state after settlement. +- Startup reconstruction must discover missing registration state in bounded + pages, without letting early failures starve later eligible endpoints. +- Empty, inactive, and indefinite-hold states must become quiet. Registered + state must remain quiet between its existing five-minute verification deadlines. + Due retries remain durable, respect Retry-After, survive restart, and retain + deferred work through drain. Join admitted provider work at shutdown. +- Verify normal and ambiguous provider outcomes through the existing protocol + tests; never replay an uncertain create merely to clear a scheduling blocker. + +## Separate remaining duties + +Runtime inventory discovery is still a separate migration. Its existing +eligibility is Discord/Photon, endpoint status verifying/active/attention, +active enabled connection, and no attention plus failed-health combination. +The shared receiver owner keeps a 15-second lease and renews an active listener +at one-third of its TTL. Those live-resource renewals are not empty discovery. +A standby must use the actual observed lease expiry for takeover, and local +ownership loss needs a callback that retains a wake after stop settlement. +Persistent transports remain host-sleep limitations even after discovery is quiet. + +`app.ts` also invokes `chatChannels.voice.reconcile()` and Teams maintenance in +this lane. Voice, Teams, milestones, wake notices, terminal confirmations, +interaction continuations, and GitHub history recovery remain distinct duties. +The worker is now wired into application startup, and runtime reconciliation no +longer initiates registration. Runtime discovery and receipt projection remain +separate. Publication and remote acceptance gates are still pending. + + +## Implemented scheduling foundation, October 25 + +The store now exposes a normalized next deadline. A genuinely unowned namespace +is immediately due; valid receipts use their existing retryAt, including the +five-minute successful verification interval. Indefinite holds, malformed records, +foreign ownership and orphaned ownership tombstones remain parked. This is only +a scheduling hint: the existing protocol still rechecks current authority and +never adopts or recreates an unowned command. The shared retry parser and history +predicate keep the reader and mutation owner consistent. + +Store-owned initial claims, receipt transitions, and retry-result updates await +transaction intent before writing. They emit endpoint-scoped hints through the +existing coordinator topic. No table or provider request was added to discovery. + +`chat-channels/discord/command-schedule.ts` owns eligible Discord endpoint +selection and maps stored state to normalized deadlines. Startup/company discovery +uses UUID keyset pages of 100. Due checks reread current eligibility and saved +deadlines before calling the existing command reconciler, then reload the saved +result. Discovery checks shutdown cancellation between pages and rows. + +`command-work.ts` uses the shared coordinator. It coalesces endpoint, connection, +or company hints; retains changes arriving during a pass; keeps passive drain +checks in memory; and dispatches at most 25 due endpoints per pass. Unsettled +attempts back off from 30 seconds to five minutes without being mistaken for +successful completion. Saved future deadlines remain authoritative. A provider +attempt already admitted at shutdown is joined through its durable settlement. + +Verification: 36 real-PostgreSQL cases pass in +`chat-discord-command-registration-store.test.ts`, and server typechecking passes. +The cases include existing ownership/ambiguity/fencing checks plus saved deadlines, +indefinite holds, empty SQL quiet, quiet between verification deadlines, startup +reconstruction, disabled eligibility, nested rollback, first committed activation +during drain, no-progress backoff, and shutdown during provider I/O. + +Two fixture failures were repaired without changing production assertions. The +foreign-owner deadline case initially violated the live-bot uniqueness constraint; +it now archives the first endpoint before creating the competing endpoint. The +empty inventory test inherited uncertain transactions from earlier fault cases; +its setup now reconciles those known prior test transactions before registering +the scoped worker. That preserves the requirement that real uncertain writes +must be reconciled before an instance can become quiet. Final evidence is in +`/tmp/sleep19-cancellation-tests.log` and `/tmp/sleep19-final-types.log` on this host. + +## Integration checklist at the foundation checkpoint + +- Wire configuration, readiness, provider failures, generic connection mutation, + connection removal, and company deletion producers to the scoped topic. Audit + all endpoint and connection writers; exempt metadata-only updates explicitly. +- Register the worker through the chat service, with existing standby, drain, + startup and shutdown admission. Preserve direct foreground setup calls. +- Remove command reconciliation from runtime reconciliation only after that + coverage exists; refresh a local runtime after successful command work through + the existing runtime owner, without reintroducing command discovery polling. +- Exercise real configuration producers and affected chat/startup consumers, then + run required workspace checks and publish the next stacked PR. Remote checks, + reviews and final reconciliation with master remain mandatory. +- Continue every remaining original inventory item and unresolved earlier PR gate. + This foundation does not finish this slice or the full goal. + + +## Integrated owner and verification + +The production wiring and producer audit are complete for this slice; see +[the behavior and producer table](../idle-discord-command-registration.md). +The shared application wiring only registers the chat-owned command worker. +Discord scheduling policy stays under `chat-channels/discord/`; shared health, +access and deletion callers use neutral chat notification contracts. The existing +runtime reconciler no longer calls the command protocol. Direct configure/resume +calls remain unchanged, and the worker refreshes an existing local runtime after +command settlement through its current owner. + +The actual configure/removal integration test proves a first configuration after +empty startup publishes its deadline, removal cancels it, and the removed worker +has no further reads or transactions. The real-protocol suite proves a committed +pause arriving during provider I/O is retained and eventually disarms the worker. +The parent health review identified company backlog starvation, and an equivalent +Discord worker regression failed before repair. Shared fair-batch selection now +gives each due company a turn, with a rotating cursor across bounded passes. +The isolated fairness helper commit is `e77979d4fc` for application to the parent. + +Local final evidence: +- `/tmp/sleep19-fairness-after.log`: 41 tests pass (38 protocol/worker + 3 selection). +- `/tmp/sleep19-discord-isolated.log`: 91 selected Discord integrations pass; + 1,327 deselected cases are not counted as passes. +- `/tmp/sleep19-connection-consumers.log`: 487 tool-access/access tests pass. +- `/tmp/sleep19-startup-consumers.log`: 20 chat coordinator/startup tests pass. +- `/tmp/sleep19-workspace-types.log` and `/tmp/sleep19-workspace-build.log`: + workspace typechecking and build pass before the final fairness addition. +- `/tmp/sleep19-fair-types.log` and `/tmp/sleep19-final-server-build.log`: final + server typechecking and build pass after the fairness addition. + +The initial integrated configuration test used a real timer before installing its +fake clock, then attempted pause while the endpoint was still verifying, and then +removed its fixture twice. Those setup/cleanup mistakes were repaired; no product +assertion was weakened. Repeated broad runs reused fixture data: inspection found +one paused endpoint for the unchanged hard-coded concurrent-identity test and two +retained contender comments for the unchanged global-body assertion. A fresh +schema-only database resolves that cross-run contamination. Dynamic registration +fixture IDs are now unique across processes, while ownership tombstones remain. + +Next gates: publish the stacked PR, complete all remote checks +and reviews, repair the health parent's fairness and producer-coverage findings, +and propagate its changes. Full original inventory and earlier unresolved CI +remain required. No merge is authorized. + + +Review at 58a1ac7595 identified a failed-item capacity regression in the shared +coordinator and a quiet-window test that leaked spies on assertion failure. The +300-entry failure-backlog test reproduced 25 attempts; the repair honors future +worker deadlines while preserving fallback/backoff and drain. The integration +fixture now waits for active-work settlement as well as an empty scheduler and +always restores spies. Verification: 87 worker/coordinator/health, eight producer +and affected Slack, 57 existing lifecycle/startup tests; server types/build pass. +The health security parent is inherited through company cleanup. See the owning +Discord document for failure evidence and exact scope. Fresh remote gates remain. + +## Idle assertion settlement repair + +A descendant CI run (#17583, run 40184498387, server shard 11/12) observed one +scoped candidate SELECT during the rollback test's quiet interval. The test only +waited for nextWakeAt to become null. The coordinator consumes a timer before +its asynchronous transaction reconciliation and scoped load run, so null alone +can mean work is running, not settled. The test now waits for zero tracked active +work as well as no deadline before measuring the same 60-second zero-SQL interval. +No production timeout, dispatch, or quiet predicate was relaxed. All 41 real +PostgreSQL command ownership cases pass with the repair. Log: +`/tmp/sleep19-settlement-tests.log`. Propagate this owner-level change through the +stack; remote checks must verify the new heads. + +## Current-parent follow-up, October 27 + +See [updated parent acceptance](../idle-discord-command-registration.md#updated-parent-acceptance-october-27) +for normal integration of the repaired health and company-cleanup parents, +127 targeted cases, corrected test selection, and the remaining remote gates. +The source duty and its 26-file scope are unchanged. + + +## Acceptance after the company-deletion parent landed, October 27 + +The branch integrates master at `f8563d25077b17681ab6ef5949f7b07415a58ede`, +including the landed health and company-deletion repairs. Squash ancestry caused +merge conflicts. Resolution applies only this slice's exact 26-file patch from +its recorded parent `21be75d2b0` to that pinned master. Every other file matches +master. The migration 0394 and current lockfile remain unchanged by this slice. + +All 141 distinct targeted cases pass: 78 command-store, batching, coordinator and +startup-timing cases; 8 real automatic Discord and held-transaction Slack cases; +and 55 complete startup/export and company-deletion cases. The company-deletion +fixture starts on an initially empty isolated database and applies migrations. +The two Discord fixtures use separate migrated database clones. The remaining +1,581 chat integration cases were deselected. Server typecheck and build pass. +The startup suite retains the existing signal-listener warnings without an +assertion failure. Source files did not change after these results; subsequent +edits record this acceptance. No full local matrix or deployed sleep is claimed. +Fresh exact-head reviews and CI remain required before merge. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 6226c069956..d62ef466d44 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -1,3 +1,4 @@ +import { deleteCompany } from "../services/company-deletion.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "../services/adapter-execution-control.js"; import { chatRunSourceMessageId } from "../services/chat-channels/run-origin.js"; import * as durableChatWakeup from "../services/durable-chat-wakeup.js"; @@ -798,11 +799,13 @@ class FakeChatSdkRuntime { const TEST_SLACK_BOT_SCOPES = "app_mentions:read,assistant:write,channels:history,channels:read,chat:write,commands,files:read,files:write,groups:history,groups:read,im:history,im:read,mpim:history,mpim:read,reactions:read,reactions:write,users:read"; +const discordApplicationPrefix = 1_400_000_000_000_000_000n + + BigInt(`0x${randomUUID().replaceAll("-", "").slice(0, 12)}`) * 1000n; let discordApplicationSequence = 0n; function uniqueDiscordApplicationId() { discordApplicationSequence += 1n; - return (123_456_789_012_345_678n + discordApplicationSequence).toString(); + return (discordApplicationPrefix + discordApplicationSequence).toString(); } function fakeSlackFetch(botId = `U-BOT-${randomUUID()}`) { @@ -7090,6 +7093,17 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...context, endpoint, callbacks }; } + async function recoverDiscordCommands(service: ChatChannelService) { + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await service.registerCommandWork(coordinator, () => true).ready; + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); + } + } + describe("Discord automatic command registration", () => { async function retireRegistrationFixture( service: ChatChannelService, @@ -7100,7 +7114,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .select() .from(chatDiscordCommandOwners) .where(eq(chatDiscordCommandOwners.endpointId, endpointId)); - await service.configure(endpointId, { action: "remove" }, "owner-user"); + if ((await service.get(endpointId))?.status !== "archived") { + await service.configure(endpointId, { action: "remove" }, "owner-user"); + } expect((await service.get(endpointId))?.status).toBe("archived"); // Removal retires only this fixture's runtime; immutable application // ownership stays available to the no-adoption/tombstone checks. @@ -7227,6 +7243,98 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("learns the first configured command and cancels its deadline when the endpoint is removed", async () => { + const f = await registrationFixture(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await f.service.registerCommandWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + await f.configure(); + expect(f.calls).toEqual(["GET", "POST"]); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now())); + const deadline = coordinator.nextWakeAt(); + expect(deadline).toBeGreaterThan(Date.now() + 200_000); + await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user"); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + expect(f.calls).toEqual(["GET", "POST"]); + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + + it.each(["credential rotation", "company deletion"] as const)("updates scheduled registration after real %s", async (mutation) => { + const f = await registrationFixture(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + let deleted = false; + let lifecycle: ReturnType | undefined; + try { + await f.service.registerCommandWork(coordinator, () => true).ready; + await f.configure(); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now())); + const before = coordinator.nextWakeAt()!; + const owners = await db.select().from(chatDiscordCommandOwners) + .where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id)); + if (mutation === "credential rotation") { + await f.service.configure(f.endpoint.id, { + action: "reconnect", credentials: { ...f.credentials, botToken: "discord-rotated-secret" }, + }, "owner-user"); + const [saved] = await db.select().from(chatActions).where(and( + eq(chatActions.endpointId, f.endpoint.id), eq(chatActions.kind, "discord_command_registration"), + )); + const expected = Date.parse(String(saved.result!.retryAt)); + expect(expected).toBeGreaterThan(before); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(expected)); + expect(f.calls).toEqual(["GET", "POST", "GET"]); + } else { + lifecycle = configureAgentLifecycle(db, { + requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete", + }); + await deleteCompany(db, f.fixture.companyId); + deleted = true; + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }); + expect(await db.select().from(chatDiscordCommandOwners) + .where(eq(chatDiscordCommandOwners.endpointId, f.endpoint.id))).toEqual(owners); + await f.service.shutdown(); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + try { + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + } finally { reads.mockRestore(); writes.mockRestore(); } + expect(f.calls).toEqual(["GET", "POST"]); + } + expect(onError).not.toHaveBeenCalled(); + } finally { + await lifecycle?.stop(); + await coordinator.stop(); vi.useRealTimers(); + if (deleted) await f.service.shutdown(); + else await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + it("keeps mention setup usable on registration failure and upgrades automatically when due", async () => { const f = await registrationFixture("unavailable"); try { @@ -7240,7 +7348,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ).toBeTypeOf("function"); f.available(); await f.makeDue(); - await f.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(f.service); expect( (await f.service.get(f.endpoint.id))?.capabilities.slashCommands, ).toBe(true); @@ -7268,6 +7376,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await f.makeDue(); restarted = createService(new FakeChatSdkRuntime(), f.providerFetch); await restarted.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(restarted.service); expect( (await restarted.service.get(f.endpoint.id))?.capabilities .slashCommands, @@ -7342,7 +7451,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; f.available(false); await f.makeDue(); - await f.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(f.service); expect(f.runtime.endpoints.get(f.endpoint.id)).toBe(original); expect(f.runtime.configurations.get(f.endpoint.id)?.callbacks).toBe( originalCallbacks, @@ -7383,7 +7492,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ); await f.makeDue(); - await f.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(f.service); expect( (await f.service.get(f.endpoint.id))?.capabilities.slashCommands, ).toBe(false); @@ -51809,6 +51918,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }); coldContexts.push(cold); await cold.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(cold.service); return { ...cold, adapter: await parser( diff --git a/server/src/app.ts b/server/src/app.ts index ad36ece1182..c80ff9e6a28 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1298,6 +1298,7 @@ export async function createApp( } registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); + chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { await chatChannels.processPublicationMaintenance(); diff --git a/server/src/services/access.ts b/server/src/services/access.ts index 45935dcdc86..f23a0ec82e5 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointWork } from "./chat-work-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -130,6 +131,7 @@ export function accessService(db: Db) { }> = []; let connectionRefs: Array<{ id: string; + transport: typeof toolConnections.$inferSelect.transport; credentialRefs: typeof toolConnections.$inferSelect.credentialRefs; credentialSecretRefs: typeof toolConnections.$inferSelect.credentialSecretRefs; }> = []; @@ -168,6 +170,7 @@ export function accessService(db: Db) { }).from(connectionGrants).where(eq(connectionGrants.companyId, companyId)), tx.select({ id: toolConnections.id, + transport: toolConnections.transport, credentialRefs: toolConnections.credentialRefs, credentialSecretRefs: toolConnections.credentialSecretRefs, }).from(toolConnections).where(eq(toolConnections.companyId, companyId)), @@ -324,6 +327,7 @@ export function accessService(db: Db) { && grant.credentialSecretRefs.length > 0 && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)), ); + if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, { companyId, id: connection.id }); await tx.update(toolConnections).set({ credentialRefs, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 5cc26141bc4..324ec43b4ae 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,7 @@ +import { notifyChatCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; +import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { notifyChatConnectionHealth } from "./chat-channels/connection-health-notifications.js"; import { credentialFingerprint } from "./chat-channels/credential-fingerprint.js"; import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; @@ -884,7 +888,15 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { authenticateGitHubWebhook } = createGitHubWebhookAuthentication(db, { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, resolveCredentials, reconcileGitHubWebhookRepository, stageGitHubWebhookIngress, processGitHubWebhookIngress, githubIngressAcceptedResponse, scheduleMessageProcessing, backgroundMessageTasks, redactError, get githubWebhookAuthenticationBarrier() { return options.githubWebhookAuthenticationBarrier; }, get deferWebhookProcessing() { return options.deferWebhookProcessing; } }); const { inspectSlackWebhookScope, recordSlackWebhookCallback } = createSlackWebhookCallbacks(db, { resolveCredentials, runtimeCallbackEndpoint, getWebhookPublicBaseUrl, scheduleMessageProcessing, slackRegistration: { notifyVerified: (endpointId) => slackRegistration.notifyVerified(endpointId) } }); - const { reconcileDiscordRuntime } = createDiscordRuntimeReconciliation(db, { reconcileDiscordCommands, endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent }); + const { reconcileDiscordRuntime } = createDiscordRuntimeReconciliation(db, { endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent }); + + function registerCommandWork(coordinator: ReturnType, canRun: () => boolean) { + return registerDiscordCommandWork(coordinator, discordCommandSchedule(db, async endpointId => { + await reconcileDiscordCommands(endpointId); + const instance = runtime.get(endpointId); + if (instance) await reconcileDiscordRuntime(endpointId, instance); + }), () => !shuttingDown && canRun()); + } async function runtimeFor( endpoint: EndpointRow, @@ -2199,6 +2211,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? "Started a conversation by mentioning the agent." : "Sent an empty message."); let comment!: Awaited>; + await notifyChatCommandRegistration(taskTx, taskEndpoint); await taskTx .update(chatEndpoints) .set({ @@ -3458,6 +3471,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerCommandWork, voice, authorizeFastResponse, slackRegistration, diff --git a/server/src/services/chat-channels/command-work-notifications.ts b/server/src/services/chat-channels/command-work-notifications.ts new file mode 100644 index 00000000000..e8f5812d4dd --- /dev/null +++ b/server/src/services/chat-channels/command-work-notifications.ts @@ -0,0 +1,18 @@ +import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; +import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; + +export async function notifyChatCommandRegistration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + if (endpoint.provider === "discord") { + await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); + } +} + +export async function notifyChatConnectionCommandRegistration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + if (transport === "chat_sdk") { + await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); + } +} + +export async function notifyChatCompanyCommandRegistration(transaction: object, companyId: string) { + await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); +} diff --git a/server/src/services/chat-channels/credentials.ts b/server/src/services/chat-channels/credentials.ts index 51ff81a807a..a2cf2b79ed2 100644 --- a/server/src/services/chat-channels/credentials.ts +++ b/server/src/services/chat-channels/credentials.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "./command-work-notifications.js"; import { prepareSlackInventory } from "./slack/inventory.js"; import { prepareGitHubInventory } from "./github/inventory.js"; import { prepareDiscordInventory } from "./discord/inventory.js"; @@ -351,6 +352,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { code: "chat_endpoint_credentials_changed", }); } + await notifyChatCommandRegistration(tx, endpoint); await tx .delete(companySecretBindings) .where( @@ -462,6 +464,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { .then((rows) => rows[0] ?? null); if (!current) throw notFound("Chat connection not found"); refs = current.refs; + await notifyChatCommandRegistration(tx, endpoint); await tx .delete(companySecretBindings) .where( diff --git a/server/src/services/chat-channels/discord/command-schedule.ts b/server/src/services/chat-channels/discord/command-schedule.ts new file mode 100644 index 00000000000..2da9f367193 --- /dev/null +++ b/server/src/services/chat-channels/discord/command-schedule.ts @@ -0,0 +1,62 @@ +import { and, eq, gt, inArray, isNotNull, ne, or } from "drizzle-orm"; +import { chatEndpoints, toolConnections, type Db } from "@paperclipai/db"; +import { readDiscordCommandRegistrationDeadline } from "../../chat-discord-command-registration-store.js"; +import type { CommandRegistrationScope } from "./command-work-notifications.js"; + +export type CommandRegistrationDeadline = { + id: string; + companyId: string; + connectionId: string; + dueAt: number; +}; + +export function discordCommandSchedule(db: Db, reconcile: (endpointId: string) => Promise) { + async function load(scope?: CommandRegistrationScope, signal?: AbortSignal): Promise { + const deadlines: CommandRegistrationDeadline[] = []; + let cursor: string | undefined; + while (true) { + signal?.throwIfAborted(); + const rows = await db.select({ + id: chatEndpoints.id, + companyId: chatEndpoints.companyId, + connectionId: chatEndpoints.connectionId, + applicationId: chatEndpoints.botExternalId, + guildId: chatEndpoints.providerAccountId, + }).from(chatEndpoints).innerJoin(toolConnections, and( + eq(toolConnections.id, chatEndpoints.connectionId), + eq(toolConnections.companyId, chatEndpoints.companyId), + )).where(and( + eq(chatEndpoints.provider, "discord"), + inArray(chatEndpoints.status, ["verifying", "active", "attention"]), + eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), + or(ne(chatEndpoints.status, "attention"), ne(toolConnections.healthStatus, "failed")), + isNotNull(chatEndpoints.botExternalId), isNotNull(chatEndpoints.providerAccountId), + scope?.kind === "endpoint" ? eq(chatEndpoints.id, scope.id) : undefined, + scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined, + scope?.kind === "company" ? eq(chatEndpoints.companyId, scope.id) : undefined, + cursor ? gt(chatEndpoints.id, cursor) : undefined, + )).orderBy(chatEndpoints.id).limit(100); + for (const row of rows) { + signal?.throwIfAborted(); + if (!row.applicationId || !row.guildId) continue; + const dueAt = await readDiscordCommandRegistrationDeadline(db, { + companyId: row.companyId, endpointId: row.id, + applicationId: row.applicationId, guildId: row.guildId, + }); + if (dueAt !== null) deadlines.push({ id: row.id, companyId: row.companyId, connectionId: row.connectionId, dueAt }); + } + if (rows.length < 100) return deadlines; + cursor = rows[rows.length - 1].id; + } + } + async function check(id: string, signal?: AbortSignal): Promise { + const scope = { kind: "endpoint", id } as const; + const [before] = await load(scope, signal); + if (!before || before.dueAt > Date.now()) return before ?? null; + signal?.throwIfAborted(); + await reconcile(id); + const [after] = await load(scope, signal); + return after ?? null; + } + return { load, check }; +} diff --git a/server/src/services/chat-channels/discord/command-work-notifications.ts b/server/src/services/chat-channels/discord/command-work-notifications.ts new file mode 100644 index 00000000000..5c8e0c78a40 --- /dev/null +++ b/server/src/services/chat-channels/discord/command-work-notifications.ts @@ -0,0 +1,10 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; + +export type CommandRegistrationScope = + | { kind: "endpoint"; id: string } + | { kind: "connection"; id: string } + | { kind: "company"; id: string }; + +export async function notifyDiscordCommandRegistration(transaction: object, scope: CommandRegistrationScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatCommandRegistration, `${scope.kind}:${scope.id}`); +} diff --git a/server/src/services/chat-channels/discord/command-work.ts b/server/src/services/chat-channels/discord/command-work.ts new file mode 100644 index 00000000000..af826d9cf1b --- /dev/null +++ b/server/src/services/chat-channels/discord/command-work.ts @@ -0,0 +1,96 @@ +import { takeCompanyWorkBatch } from "../../company-work-batch.js"; +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import type { CommandRegistrationDeadline, discordCommandSchedule } from "./command-schedule.js"; +import type { CommandRegistrationScope } from "./command-work-notifications.js"; + +function decodeScope(hint: string): CommandRegistrationScope | undefined { + if (hint === "*") return undefined; + const separator = hint.indexOf(":"); + const kind = hint.slice(0, separator), id = hint.slice(separator + 1); + if (kind !== "endpoint" && kind !== "connection" && kind !== "company") + throw new Error("Unknown command registration scope"); + return { kind, id }; +} + +function matches(entry: CommandRegistrationDeadline, scope?: CommandRegistrationScope) { + if (!scope) return true; + return (scope.kind === "endpoint" ? entry.id : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id; +} + +export function registerDiscordCommandWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const deadlines = new Map(); + const failures = new Map(); + const retryMs = 30_000; + let earliest = 0; + let lastCompany: string | null = null; + function next() { + let at: number | null = dirty.size ? earliest : null; + for (const entry of deadlines.values()) { + const due = Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0); + at = at === null ? due : Math.min(at, due); + } + return at; + } + return coordinator.register(DELIVERY_QUEUES.chatCommandRegistration, { + retryMs, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + retryMs; + const requested = dirty.has("*") ? ["*"] : [...dirty]; + dirty.clear(); + const errors: unknown[] = []; + const unavailable: Array = []; + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { + for (const remaining of requested.slice(index)) dirty.add(remaining); + return; + } + const hint = requested[index]; + let scope: CommandRegistrationScope | undefined; + try { + scope = decodeScope(hint); + const rows = await service.load(scope, signal); + for (const entry of deadlines.values()) if (matches(entry, scope)) deadlines.delete(entry.id); + for (const entry of rows) deadlines.set(entry.id, entry); + for (const id of failures.keys()) { + const entry = deadlines.get(id); + if (!entry || entry.dueAt > Date.now()) failures.delete(id); + } + } catch (error) { dirty.add(scope ? hint : "*"); unavailable.push(scope); errors.push(error); } + } + const due = [...deadlines.values()] + .filter(entry => !unavailable.some(scope => matches(entry, scope)) && + Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)); + const batch = takeCompanyWorkBatch(due, 25, lastCompany); + lastCompany = batch.lastCompany; + for (const entry of batch.entries) { + if (signal.aborted || !canRun()) break; + try { + const current = await service.check(entry.id, signal); + if (current) deadlines.set(entry.id, current); else deadlines.delete(entry.id); + if (current && current.dueAt <= Date.now()) throw new Error("Discord command registration remains due"); + failures.delete(entry.id); + } catch (error) { + const count = Math.min((failures.get(entry.id)?.count ?? 0) + 1, 5); + failures.set(entry.id, { count, retryAt: Date.now() + Math.min(300_000, retryMs * 2 ** (count - 1)) }); + errors.push(error); + } + } + if (due.length > batch.entries.length) earliest = Date.now() + 1_000; + if (errors.length) throw new AggregateError(errors, "Discord command registration could not settle"); + }, + // Preserve hints/deadlines for resume without holding settled intent during drain. + // The coordinator independently fences unresolved database transactions. + hasPending: async () => canRun() && dirty.size > 0, + nextRunAt: next, + }); +} diff --git a/server/src/services/chat-channels/discord/gateway-events.ts b/server/src/services/chat-channels/discord/gateway-events.ts index f21c78cc11f..47deeff7941 100644 --- a/server/src/services/chat-channels/discord/gateway-events.ts +++ b/server/src/services/chat-channels/discord/gateway-events.ts @@ -1,3 +1,4 @@ +import { notifyDiscordCommandRegistration } from "./command-work-notifications.js"; import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence, InternalSetupState } from "../runtime-types.js"; import { notifyChatEndpointWork } from "../../chat-work-notifications.js"; @@ -129,6 +130,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents } const now = new Date(); if (event.type === "connecting") { + if (current.status === "attention") await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); await tx .update(chatEndpoints) .set({ @@ -157,6 +159,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents if (current.status === "attention" && !gatewayOwnedAttention) return; const status = current.setup.step === "complete" ? "active" : "verifying"; + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId); await tx .update(chatEndpoints) @@ -186,6 +189,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents return; } + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); const detail = discordGatewayFailureDetail(event); const transientGuildOutage = event.type === "guild_unavailable"; const terminal = diff --git a/server/src/services/chat-channels/discord/runtime-reconciliation.ts b/server/src/services/chat-channels/discord/runtime-reconciliation.ts index 8a59d6576b2..bd2305ae72f 100644 --- a/server/src/services/chat-channels/discord/runtime-reconciliation.ts +++ b/server/src/services/chat-channels/discord/runtime-reconciliation.ts @@ -7,11 +7,9 @@ import type { Db } from "@paperclipai/db"; import type { EndpointRow, EndpointRecord, RuntimeContext } from "../runtime-types.js"; import type { ChatRuntime } from "../../chat-runtime.js"; import type { ChatEndpointRuntime } from "../../voice/voice-runtime.js"; -import type { createDiscordCommands } from "./commands.js"; import type { createChatMessageLifecycle } from "../inbound/lifecycle.js"; import type { ToolCredentialSecretRef } from "@paperclipai/shared"; type ReconciliationOptions = { - reconcileDiscordCommands: ReturnType["reconcileDiscordCommands"]; endpointRecord: (endpointId: string) => Promise; runtimeContexts: WeakMap; invalidateRuntime: (endpointId: string) => Promise; @@ -22,9 +20,8 @@ type ReconciliationOptions = { discordGatewayRuntimeIsCurrent: (endpointId: string, context: RuntimeContext) => boolean; }; export function createDiscordRuntimeReconciliation(db: Db, options: ReconciliationOptions) { - const { reconcileDiscordCommands, endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent } = options; + const { endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent } = options; async function reconcileDiscordRuntime(endpointId: string, instance: ChatEndpointRuntime) { - await reconcileDiscordCommands(endpointId); const afterCommands = await endpointRecord(endpointId); const desiredCommands = afterCommands?.endpoint.capabilities.slashCommands && diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 8fe0f601e36..49e4f014a8e 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "../command-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { slackQuestionFormValidationError } from "../slack/question-forms.js"; import { chatInteractionForm, chatItemVerdicts } from "../../chat-interaction-forms.js"; @@ -417,6 +418,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb ["active"], ); if (currentEndpoint) { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/provider-effects.ts b/server/src/services/chat-channels/provider-effects.ts index 4b15a1b769d..1273c07c59b 100644 --- a/server/src/services/chat-channels/provider-effects.ts +++ b/server/src/services/chat-channels/provider-effects.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "./command-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js"; @@ -317,6 +318,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) { ); } if (input.disposition.kind === "endpoint_attention" && currentEndpoint) { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index 84184bb8633..ae2580dfb3f 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "../command-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; @@ -189,6 +190,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur : null; if (disposition.kind === "endpoint_attention" && currentEndpoint) { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index a706452196e..54cba206d58 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "./command-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import type { ConnectionManagementContext } from "../connection-management.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; @@ -510,6 +511,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc : currentEndpoint.status === "attention" ? "verifying" : currentEndpoint.status; + await notifyChatCommandRegistration(tx, currentEndpoint); if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId); await tx .update(chatEndpoints) @@ -539,6 +541,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc .where(eq(toolConnections.id, currentEndpoint.connectionId)); } } else { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-discord-command-registration-store.test.ts b/server/src/services/chat-discord-command-registration-store.test.ts index 02f8c877475..44809957ca4 100644 --- a/server/src/services/chat-discord-command-registration-store.test.ts +++ b/server/src/services/chat-discord-command-registration-store.test.ts @@ -1,3 +1,10 @@ +import { reconcileDatabaseWork } from "@paperclipai/db"; +import { idleWorkSnapshot } from "./task-admission.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; +import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; +import { notifyDiscordCommandRegistration } from "./chat-channels/discord/command-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { createHash, randomUUID } from "node:crypto"; import { afterAll, @@ -23,6 +30,7 @@ import { } from "@paperclipai/db"; import { discordPaperclipCommandDefinition } from "./chat-discord-command-registration.js"; import { + readDiscordCommandRegistrationDeadline, readRegisteredDiscordCommandRegistration, reconcileStoredDiscordCommandRegistration, type StoredDiscordCommandRegistrationOptions, @@ -55,7 +63,33 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => { await db?.$client.end(); await temporary?.cleanup(); }); - afterEach(() => vi.restoreAllMocks()); + let coordinator: ReturnType | undefined; + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + }); + async function scheduled(companyId: string, reconcile: (id: string) => Promise, canRun = () => true) { + await Promise.all(Array.from({ length: 4 }, () => db.execute(sql`select 1`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + const service = discordCommandSchedule(db, reconcile); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile: () => true, onError: error => errors.push(error) }); + const worker = registerDiscordCommandWork(coordinator, { + ...service, + load: (scope, signal) => service.load(scope ?? { kind: "company", id: companyId }, signal), + }, canRun); + await worker.ready; + return { service, errors }; + } + async function sqlQuiet(milliseconds: number) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(milliseconds); + expect(reads).not.toHaveBeenCalled(); + expect(writes).not.toHaveBeenCalled(); + expect(execute).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); execute.mockRestore(); + } async function fixture(applicationId = appId()) { const companyId = randomUUID(), @@ -628,4 +662,256 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => { expect(f.calls).toEqual(["GET", "POST"]); }, ); + it("exposes saved verification and retry deadlines without contacting the provider", async () => { + const f = await fixture(); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(0); + expect(f.calls).toEqual([]); + expect(await f.reconcile()).toMatchObject({ kind: "registered" }); + const row = await f.read(); + const due = Date.parse(String(row!.result!.retryAt)); + expect(due).toBeGreaterThan(Date.now() + 290_000); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(due); + expect(f.calls).toEqual(["GET", "POST"]); + const retriedAt = new Date(Date.now() + 601_125).toISOString(); + await db.update(chatActions).set({ result: { ...row!.result, outcome: "unavailable", retryAt: retriedAt } }) + .where(eq(chatActions.id, row!.id)); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(Date.parse(retriedAt)); + expect(f.calls).toEqual(["GET", "POST"]); + }); + it("parks indefinite and malformed retry records without inventing an earlier deadline", async () => { + const f = await fixture(); await f.reconcile(); const row = (await f.read())!; + for (const result of [ + { ...row.result, retryAt: "+275760-09-13T00:00:00.000Z", retryIndefinite: true }, + { ...row.result, retryAt: "invalid" }, + { ...row.result, schema: "unknown" }, + ]) { + await db.update(chatActions).set({ result }).where(eq(chatActions.id, row.id)); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBeNull(); + } + expect(f.calls).toEqual(["GET", "POST"]); + }); + it("does not schedule ownership tombstones or another endpoint's registration", async () => { + const f = await fixture(); await f.reconcile(); + await db.update(chatEndpoints).set({ status: "archived" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + const other = await fixture(f.scope.applicationId); + expect(await readDiscordCommandRegistrationDeadline(db, other.scope)).toBeNull(); + await db.delete(chatActions).where(eq(chatActions.id, (await f.read())!.id)); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBeNull(); + expect(await readDiscordCommandRegistrationDeadline(db, other.scope)).toBeNull(); + expect(other.calls).toEqual([]); + }); + it("retains the stored retry deadline for an uncertain provider write", async () => { + const f = await fixture(); + f.beforeResponse(async method => { if (method === "POST") throw new Error("lost response"); }); + expect(await f.reconcile()).toMatchObject({ kind: "unknown" }); + const row = (await f.read())!; + expect(row.payload.registration).toMatchObject({ phase: "attempted" }); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(Date.parse(String(row.result!.retryAt))); + expect(f.calls).toEqual(["GET", "POST"]); + }); + + it("leaves no deadline or SQL after an empty registration inventory", async () => { + const reconcile = vi.fn(); + const { errors } = await scheduled(randomUUID(), reconcile); + expect(coordinator!.nextWakeAt()).toBeNull(); + await sqlQuiet(180_000); + expect(reconcile).not.toHaveBeenCalled(); expect(errors).toEqual([]); + }); + it("reconstructs a registered verification deadline and stays quiet until it is due", async () => { + const f = await fixture(); await f.reconcile(); + const due = Date.parse(String((await f.read())!.result!.retryAt)); + const { errors } = await scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + expect(coordinator!.nextWakeAt()).toBe(due); + await sqlQuiet(120_000); + expect(f.calls).toEqual(["GET", "POST"]); + await vi.advanceTimersByTimeAsync(due - Date.now()); + await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST", "GET"])); + await vi.waitFor(async () => expect(Date.parse(String((await f.read())!.result!.retryAt))).toBeGreaterThan(due)); + expect(errors).toEqual([]); + }); + it("keeps indefinite holds quiet and respects disabled endpoint eligibility", async () => { + const f = await fixture(); await f.reconcile(); const row = (await f.read())!; + await db.update(chatActions).set({ result: { ...row.result, retryIndefinite: true } }).where(eq(chatActions.id, row.id)); + const reconcile = vi.fn(); + const { service, errors } = await scheduled(f.scope.companyId, reconcile); + await sqlQuiet(600_000); expect(coordinator!.nextWakeAt()).toBeNull(); + await db.update(chatActions).set({ result: { ...row.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, row.id)); + await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + expect(await service.check(f.scope.endpointId)).toBeNull(); + expect(reconcile).not.toHaveBeenCalled(); expect(errors).toEqual([]); + }); + it("honors outer rollback and wakes from committed endpoint changes after drain", async () => { + const f = await fixture(); + await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + let draining = false; + const { errors } = await scheduled(f.scope.companyId, async () => { await f.reconcile(); }, () => !draining); + const activate = async (tx: typeof db) => { + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: f.scope.endpointId }); + await tx.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + }; + await expect(db.transaction(async tx => { + await tx.transaction(async nested => activate(nested as unknown as typeof db)); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.calls).toEqual([]); + throw new Error("abort"); + })).rejects.toThrow("abort"); + await vi.advanceTimersByTimeAsync(60_000); + // A running sweep has already consumed its timer. Wait for its SQL and + // transaction reconciliation to settle before measuring idle traffic. + await vi.waitFor(() => { + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }); + await sqlQuiet(60_000); + draining = true; + await db.transaction(async tx => activate(tx as unknown as typeof db)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); }); + const drained = idleWorkSnapshot(); + await sqlQuiet(90_000); expect(idleWorkSnapshot()).toEqual(drained); expect(f.calls).toEqual([]); + draining = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST"])); + expect(errors).toEqual([]); + }); + it("rechecks persisted eligibility before a due provider call", async () => { + const f = await fixture(); await f.reconcile(); + const { service, errors } = await scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + await f.due(); + await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, + (await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.scope.endpointId)))[0].connectionId)); + expect(await service.check(f.scope.endpointId)).toBeNull(); + expect(f.calls).toEqual(["GET", "POST"]); expect(errors).toEqual([]); + }); + + it("backs off an attempt that cannot advance its durable state", async () => { + const f = await fixture(); + const reconcile = vi.fn(async () => {}); + const { errors } = await scheduled(f.scope.companyId, reconcile); + expect(reconcile).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(reconcile).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(errors).toHaveLength(2)); + await sqlQuiet(30_000); + expect(reconcile).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(reconcile).toHaveBeenCalledTimes(3)); + expect(await f.read()).toBeUndefined(); + }); + it("joins an admitted provider attempt during shutdown", async () => { + const f = await fixture(); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + f.beforeResponse(async method => { if (method === "GET") await held; }); + const starting = scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + await vi.waitFor(() => expect(f.calls).toEqual(["GET"])); + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await Promise.resolve(); expect(stopped).toBe(false); + release(); await stopping; const { errors } = await starting; + expect(stopped).toBe(true); + expect(await f.read()).toMatchObject({ status: "processed" }); + expect(coordinator!.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); + }); + + it("retains a committed eligibility change while the provider attempt is in flight", async () => { + const f = await fixture(); await f.reconcile(); await f.due(); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + f.beforeResponse(async method => { if (method === "GET") await held; }); + const starting = scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + try { + await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST", "GET"])); + await db.transaction(async tx => { + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: f.scope.endpointId }); + await tx.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + }); + } finally { release(); } + const { errors } = await starting; + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await sqlQuiet(120_000); + expect(f.calls).toEqual(["GET", "POST", "GET"]); expect(errors).toEqual([]); + }); + + it("checks a quiet company's due command in the first batch despite another company's backlog", async () => { + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const entries = Array.from({ length: 30 }, (_, index) => ({ + id: `busy-${index.toString().padStart(2, "0")}`, companyId: "busy", connectionId: "busy-connection", dueAt: 0, + })); + entries.push({ id: "quiet", companyId: "quiet", connectionId: "quiet-connection", dueAt: 1 }); + const check = vi.fn(async (id: string) => ({ ...entries.find(entry => entry.id === id)!, dueAt: Date.now() + 300_000 })); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerDiscordCommandWork(coordinator, { load: async () => entries, check }, () => true).ready; + expect(check).toHaveBeenCalledTimes(25); + expect(check.mock.calls.map(call => call[0])).toContain("quiet"); + expect(errors).toEqual([]); + }); + + it("drains a large due backlog without a thirty-second delay between bounded batches", async () => { + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const entries = Array.from({ length: 300 }, (_, index) => ({ + id: `command-${index}`, companyId: `company-${index % 10}`, connectionId: `connection-${index}`, dueAt: 0, + })); + const check = vi.fn(async (id: string) => ({ ...entries.find(entry => entry.id === id)!, dueAt: Date.now() + 300_000 })); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerDiscordCommandWork(coordinator, { load: async () => entries, check }, () => true).ready; + expect(check).toHaveBeenCalledTimes(25); + await vi.advanceTimersByTimeAsync(15_000); + expect(check).toHaveBeenCalledTimes(300); + expect(new Set(check.mock.calls.map(([id]) => id)).size).toBe(300); + await sqlQuiet(120_000); + expect(check).toHaveBeenCalledTimes(300); + expect(errors).toEqual([]); + }); + + it("drains healthy commands promptly while one entry remains in failure backoff", async () => { + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const entries = Array.from({ length: 300 }, (_, index) => ({ + id: `command-${index}`, companyId: `company-${index % 10}`, connectionId: `connection-${index}`, dueAt: 0, + })); + const check = vi.fn(async (id: string) => { + if (id === "command-0") throw new Error("Unavailable credential"); + return { ...entries.find(entry => entry.id === id)!, dueAt: Date.now() + 300_000 }; + }); + const onError = vi.fn(); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + await registerDiscordCommandWork(coordinator, { load: async () => entries, check }, () => true).ready; + await vi.advanceTimersByTimeAsync(15_000); + expect(new Set(check.mock.calls.map(([id]) => id)).size).toBe(300); + expect(check.mock.calls.filter(([id]) => id === "command-0")).toHaveLength(1); + expect(onError).toHaveBeenCalledOnce(); + await vi.advanceTimersByTimeAsync(45_000); + expect(check.mock.calls.filter(([id]) => id === "command-0")).toHaveLength(2); + expect(check.mock.calls.filter(([id]) => id !== "command-0")).toHaveLength(299); + }); + it("recovers queued scopes after an unknown hint instead of dropping the wake", async () => { + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const entry = { id: "endpoint", companyId: "company", connectionId: "connection", dueAt: 0 }; + let owed = false; + const check = vi.fn(async () => { owed = false; return null; }); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerDiscordCommandWork(coordinator, { load: async () => owed ? [entry] : [], check }, () => true).ready; + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatCommandRegistration, "unknown:legacy"); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: entry.id }); + owed = true; + }); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1_000); + expect(errors).toHaveLength(1); + }, { timeout: 5_000 }); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(check).toHaveBeenCalledTimes(1)); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await sqlQuiet(120_000); + expect(errors).toHaveLength(1); + }); + }); diff --git a/server/src/services/chat-discord-command-registration-store.ts b/server/src/services/chat-discord-command-registration-store.ts index 22123da3bbe..749b2d93e5c 100644 --- a/server/src/services/chat-discord-command-registration-store.ts +++ b/server/src/services/chat-discord-command-registration-store.ts @@ -1,3 +1,4 @@ +import { notifyDiscordCommandRegistration } from "./chat-channels/discord/command-work-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { @@ -66,6 +67,19 @@ function safeResult(outcome: string, seconds?: number) { return { schema: resultSchema, outcome, ...retrySchedule(outcome, seconds) }; } +function registrationRetryAt(result: typeof chatActions.$inferSelect["result"]): number | null { + if (result?.schema !== resultSchema || typeof result.retryAt !== "string") return null; + const at = Date.parse(result.retryAt); + return Number.isFinite(at) ? at : null; +} + +async function hasRegistrationHistory(database: Database, applicationId: string): Promise { + const [history] = await database.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.kind, kind), eq(chatActions.providerActionId, actionKey(applicationId)), + )).limit(1); + return history !== undefined; +} + async function stored( database: Database, scope: DiscordCommandRegistrationScope, @@ -136,6 +150,21 @@ export async function readRegisteredDiscordCommandRegistration( : null; } +export async function readDiscordCommandRegistrationDeadline( + database: Database, + scope: DiscordCommandRegistrationScope, +): Promise { + const current = await stored(database, scope, false, true); + if (current) { + if (current.row.result?.retryIndefinite === true) return null; + return registrationRetryAt(current.row.result); + } + const [owner] = await database.select({ applicationId: chatDiscordCommandOwners.applicationId }) + .from(chatDiscordCommandOwners).where(eq(chatDiscordCommandOwners.applicationId, scope.applicationId)); + if (owner || await hasRegistrationHistory(database, scope.applicationId)) return null; + return 0; +} + export async function reconcileStoredDiscordCommandRegistration( db: Db, options: StoredDiscordCommandRegistrationOptions, @@ -169,17 +198,8 @@ export async function reconcileStoredDiscordCommandRegistration( .where(eq(chatDiscordCommandOwners.applicationId, scope.applicationId)) .for("update"); if (!priorOwner) { - const [history] = await tx - .select({ id: chatActions.id }) - .from(chatActions) - .where( - and( - eq(chatActions.kind, kind), - eq(chatActions.providerActionId, actionKey(scope.applicationId)), - ), - ) - .limit(1); - if (history) deny(); // No adoption of history lacking the global claim. + if (await hasRegistrationHistory(tx, scope.applicationId)) deny(); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: scope.endpointId }); const actionId = randomUUID(); const inserted = await tx .insert(chatDiscordCommandOwners) @@ -209,11 +229,8 @@ export async function reconcileStoredDiscordCommandRegistration( const current = await stored(tx, scope, true, true); if (!current) deny(); // In particular, never recreate an orphaned action. const result = current.row.result; - const due = - result?.schema === resultSchema && typeof result.retryAt === "string" - ? Date.parse(result.retryAt) - : NaN; - if (!Number.isFinite(due)) deny(); + const due = registrationRetryAt(result); + if (due === null) deny(); return { state: current.state, deferred: @@ -253,6 +270,7 @@ export async function reconcileStoredDiscordCommandRegistration( await db.transaction(async (tx) => { const current = await check(tx); if (!same(current.state, expected)) deny(); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: scope.endpointId }); const result = safeResult( next.phase === "registered" ? "registered" : "unknown", ); @@ -272,6 +290,7 @@ export async function reconcileStoredDiscordCommandRegistration( }); await db.transaction(async (tx) => { const current = await check(tx); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: scope.endpointId }); await tx .update(chatActions) .set({ diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 1d8b1ffa518..cc4742d0b71 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionCommandRegistration } from "./chat-channels/command-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export async function notifyChatDeliveryWork(tx: object): Promise { @@ -37,4 +38,5 @@ export async function notifyChatEndpointWork(tx: object, connectionId: string): for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue); await notifyChatVerificationWork(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTelegramSubscriptions, connectionId); + await notifyChatConnectionCommandRegistration(tx, connectionId); } diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index ca79c6a4271..c87d7f89a1d 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyChatCompanyCommandRegistration } from "./chat-channels/command-work-notifications.js"; import { chatCompanyDeletion } from "./chat-channels/company-deletion.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; @@ -64,6 +65,7 @@ export async function deleteCompany(db: Db, id: string) { const [existing] = await tx.select({ id: companies.id }).from(companies) .where(eq(companies.id, id)).for("no key update"); if (!existing) return null; + await notifyChatCompanyCommandRegistration(tx, id); await chatCompanyDeletion.deleteCompanyData(tx, id); await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync); await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 889d566681a..e95a08eb41e 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -43,6 +43,7 @@ export const DELIVERY_QUEUES = { chatGitHubPublications: "chat-github-publications", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", + chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs", chatVerification: "chat-verification", diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index f22291cd81c..0ea2ab9203a 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionCommandRegistration } from "../../chat-channels/command-work-notifications.js"; import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { connectionActorUserId } from "../actor.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; @@ -48,8 +49,9 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d message: string | null, ) { const now = new Date(); - const [updated] = await writeToolConnectionHealthChange(db, connection, tx => tx - .update(toolConnections) + const [updated] = await writeToolConnectionHealthChange(db, connection, async tx => { + await notifyChatConnectionCommandRegistration(tx, connection.id, connection.transport); + return tx.update(toolConnections) .set({ healthStatus: status, healthMessage: message, @@ -59,7 +61,8 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d updatedAt: now, }) .where(eq(toolConnections.id, connection.id)) - .returning()); + .returning(); + }); if (connection.transport === "local_stdio") { await db .update(toolRuntimeSlots) diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index 3a8b7073275..6c4c61f4941 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointWork } from "../../chat-work-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { toConnection } from "./projection.js"; import { actorBinding } from "../actor.js"; @@ -539,6 +540,7 @@ async function removeConnection( .returning({ id: connectionTokenIssuances.id }); const archived = await db.transaction(async (tx) => { + if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); await notifyToolConnectionHealth(tx, connection); await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId)))