Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-10 12:37:55 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix(server): wake chat receiver discovery from configuration and lease deadlines (#17572)

Browse Source 
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Chat receivers own persistent provider connections and fenced
database leases.
> - The chat maintenance timer repeatedly queries all eligible
receivers, including when none exist.
> - Configuration changes and known lease deadlines already identify
when discovery is needed.
> - This pull request moves receiver discovery to the shared delivery
coordinator.
> - Empty or settled discovery stops querying, while ownership and
recovery remain explicit.

## Linked Issues or Issue Description

Refs #17557. This is the next stacked slice after Discord command
registration. The base is `codex/sleep-19-discord-command-registration`.

**What happened?**
The recurring chat maintenance lane scans receiver inventory and
reconciles Discord and Photon runtimes even when configuration has not
changed.

**Expected behavior**
Reconstruct receiver work at startup. React to committed configuration
changes and local listener retirement. Retry foreign ownership at the
saved lease expiry. Stop discovery when there is no remaining
obligation.

**Steps to reproduce**
Start an instance with no chat endpoints. Let startup finish. Observe
the recurring receiver inventory query from chat maintenance. With a
configured receiver, remove its ownership token while listener shutdown
is delayed and signal a configuration refresh: the old implementation
can replace the listener before shutdown completes.

## What Changed

- Preserve healthy receiver progress after a slow failed batch while
retaining per-receiver retry backoff.

- Register a receiver discovery worker with scoped refreshes, paged
startup inventory, fair batches, and per-receiver or scope failure
backoff.
- Keep owned receiver identities cached without a discovery deadline, so
deletion still stops their listeners.
- Use existing transaction signals for configuration, credentials,
health, access revocation, company deletion, and provider lifecycle
changes. Rename the generic command-only helper to describe its broader
configuration contract.
- Keep lease-key selection with the receiver owner. Shared scheduling
contains no provider protocol.
- Retain stopping ownership until listener shutdown and lease release
complete. Concurrent acquisition joins the existing stop promise.
- Remove receiver discovery from the periodic chat lane. Preserve voice
and the other unmigrated duties.
- Record producer coverage, failure evidence, limits, and remaining work
in `doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md`.

## Verification

Published head: `ab39569250f4b699a0ab7a829e66dbe27c15020b`. Discord
#17557 has landed at master `5bf0142d45`. A normal merge reconciles its
squash ancestry by applying only the recorded 27-file receiver patch
from parent `547fdf2139`. Every other path matches that master. Every
server and package production/test file is byte-identical to the
published `6e88ca0201` source that passed the 131-case run, typecheck
and build. Unrelated upstream UI/eval/documentation changes are
preserved. This PR targets master. Fresh exact-head reviews and CI are
required.

- **131 distinct targeted cases pass.** Five complete receiver,
coordinator, batching, startup, and GitHub health error-projection
suites pass 99 cases. Twelve real Discord
automatic-registration/receiver and Slack held-transaction cases pass
separately, with 1,581 other chat integration cases deselected. The
complete Photon channel suite passes 20 cases.
- The standalone receiver and Photon fixtures each reproduced
missing-table failures on separate initially empty PostgreSQL 17
databases. Both now apply pending migrations before client use; their
complete 20-case files pass on those same databases. The large chat
producer suite uses a separate migrated clone.
- The old review concerns already have source fixes: discovery stops
suppress their own completion wake at `1604a9fcba`; slow failed batches
supply `nextRunAtOnError` at `ab56393acc`. Guarded mutations prove the
existing regressions: reenabling the stop wake yields two inventory
reads instead of one, and removing the error deadline fails the 1,100 ms
provider-delay next-wake assertion. Both original source files are
restored byte for byte. Normal full worker and real producer suites pass
with those fixes.
- All 131 distinct targeted cases, production server typecheck and build
pass on source `6e88ca0201`. Exact source comparison carries that
evidence to this normal merge; the unchanged suites were not rerun just
for unrelated upstream changes. Receiver and Photon fixtures
self-migrate separate initially empty databases; the large producer
fixture uses its own migrated clone. Source is unchanged after these
checks. Recorded listener-count warnings and controlled provider failure
logs remain; no assertion was weakened. An initial whole-file identity
check rejected expected parent queue additions; precise patch comparison
confirms those declarations are correctly preserved.
- Old server shard 12 failed because GitHub reports that its self-hosted
runner lost communication. The grouped-test step is cancelled, and the
retained log has no failing assertion or completed summary. The
underlying host/resource/network cause is unproven. That interrupted job
is not a pass; the updated parent and fixture source require a fresh
full CI run.
- [Current acceptance and failure
record](https://github.com/paperclipai/paperclip/blob/ab39569250f4b699a0ab7a829e66dbe27c15020b/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md#compatibility-after-discord-squash-merge-october-27).
Prior 218-case and full-workspace results describe earlier revisions. No
full local repository-suite, multi-writer deployment, Cloud wake, host
suspension, or fleet cost claim is made.

## Risks

- Process-local transaction hints cover the active database owner and
its shared clients. Arbitrary direct SQL or independent configuration
writers need explicit wake integration. Startup/resume reconstruction
and known foreign-lease takeover remain supported; this does not add
general multi-writer discovery.
- Active receiver sockets and lease renewals continue. This change does
not prove host suspension or extend Cloud alarm coverage.
- Configuration signals can arrive during drain, initialization, or
shutdown. The worker preserves dirty scopes and the owner joins listener
retirement before replacement.
- Voice, Teams, milestone projection, publication maintenance, GitHub
recovery, and the remaining stack inventory are separate work.
- The Discord parent has landed and the receiver child now targets
master. Fresh exact-head reviews and all CI must pass before merge.

## Model Used

OpenAI GPT-6 via Codex, with repository inspection, code execution, and
test tools. The runtime does not expose a more specific deployment
identifier or context-window limit.

## Checklist

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

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-10-28 07:25:20 -05:00
1 parent 7d8bab773d
commit dce0042593
27 files changed
+1123 -73
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

doc/plans/2026-10-24-recurring-server-sleep-stack.md
+55 -1
View File
@@ -17,7 +17,7 @@ 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 | #17554 repaired at `8c1d293d1a`; fresh CI/review in progress |
| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | [#17554](https://github.com/paperclipai/paperclip/pull/17554) merged after all checks and both 5/5 reviews |
| 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 |
@@ -1483,6 +1483,42 @@ 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.
## Slice 20 receiver implementation
Receiver discovery now uses scoped committed configuration changes, startup
reconstruction, foreign-lease deadlines, and listener-stop completion. Voice and
all other maintenance reasons remain separate. See
[the receiver audit](2026-10-25-chat-receiver-discovery-sleep-audit.md).
A failed-scope deadline regression and a real concurrent listener-stop race were
reproduced and repaired. All 218 selected receiver/startup, Discord, Photon,
command/delivery and standby cases pass, with workspace typechecking passing.
Workspace build also passes; publication and fresh remote gates remain open. This is not
completion of the slice or the full stack. Master is refreshed at 0d73505098;
final stack/master reconciliation remains required.
Parent Discord #17557 at d3187f76a4 and company cleanup #17561 at e8febb30eb now
have no failed or pending checks (Paperclip Review skipped). No merge was made.
The older #17490 E2E cancellation failure remains unresolved. The entire original
inventory stays in scope, with voice the next separate reason after receivers.
## Receiver review repair
The real company-removal producer regression reproduced two inventory reads for
one observed removal. Discovery-initiated stops now suppress their own completion
wake; independent token loss still wakes discovery after listener retirement.
A full reconstruction also covers its already-pending scoped hints. Hints that
arrive during the load remain dirty, and narrower writes cannot bypass a failed
full reconstruction's retry deadline.
Both redundant-load regressions failed before repair. All 45 selected receiver,
startup and real Discord producer tests pass after repair, including the added
full-refresh failure/backoff case. Server typecheck and build pass. Prior broad
consumer/workspace verification remains recorded above; these 45 overlap that
coverage and are not an additional unique total. Publication requires fresh
remote CI and review. The full migration inventory remains outstanding.
## Acceptance follow-up after the plugin parent merged (October 27)
Eighteen preceding sleep PRs are merged through plugin scheduling #17518 at
@@ -1498,6 +1534,16 @@ 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.
## Receiver follow-up with current Discord parent, October 27
Nineteen sleep PRs are merged through health #17554. Receiver discovery #17572
now includes current Discord parent `cedf2030c4`, retaining the original 27-file
scope. Its existing slow-failure and stop-wake repairs have guarded mutation
proofs, 131 targeted cases pass, and both newly supported external fixtures
self-migrate. See [the owning audit](2026-10-25-chat-receiver-discovery-sleep-audit.md#current-parent-and-review-acceptance-october-27)
for exact evidence and the older CI runner-loss failure. Fresh CI/reviews and
parent landing are still required. The original remaining-duty inventory is unchanged.
## Discord registration landing compatibility, October 27
@@ -1505,3 +1551,11 @@ 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.
## Receiver integration with landed foundations, October 27
Receiver discovery keeps its 27-file scope on Discord parent `547fdf2139` and
preserves current master queue declarations. All 131 targeted cases, server
typecheck and build pass. See the [receiver audit](2026-10-25-chat-receiver-discovery-sleep-audit.md#landed-foundation-integration-acceptance-october-27).
Fresh exact-head review, CI and parent landing remain required.
doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md
+351
View File
@@ -0,0 +1,351 @@
# Chat receiver discovery sleep migration audit
Parent: Discord command registration #17557, head d3187f76a4.
Master refreshed during this continuation: e6ec781a1a.
Branch: codex/sleep-20-chat-runtime-discovery.
Status: implementation and focused lifecycle verification; publication and final acceptance pending.
## One reason for this PR
Remove recurring inventory discovery for leased chat receivers from the aggregate
chat maintenance timer. Preserve receipt projection, credential fencing, runtime
initialization, takeover, and stop semantics. Discord command registration belongs
to the parent; voice reconciliation, publication maintenance, milestones, Teams,
and GitHub recovery remain separate migration reasons.
`server/src/app.ts` currently invokes `chatChannels.reconcileProviderRuntimes()`
and `chatChannels.voice.reconcile()` in the same recurring maintenance lane.
The receiver method in `services/chat-channels.ts` queries every eligible Discord
or iMessage Photon endpoint, stops local ownership missing from that inventory,
then calls `runtimeFor` and `reconcileDiscordRuntime` for each eligible endpoint.
This query runs even when there are no receivers. Separating voice is necessary
before removing this callback from the maintenance lane; removing both would lose
an unmigrated obligation.
## Durable truth and owner boundaries
Existing `chat_endpoints`, `tool_connections`, and `chat_endpoint_leases` contain
the necessary configuration and ownership. No additional durable queue table is
needed. The receiver owner must retain the current eligibility policy: supported
leased provider; verifying/active/attention endpoint; active, enabled connection;
and exclusion of attention plus failed connection health.
The receiver provider policy currently lives in `leasedChatProvider` in
`chat-channels.ts` and the lease-key selection inside `runtime/ownership.ts`.
Centralize that existing policy with the chat provider/receiver owner before
adding another independent Discord/Photon switch in scheduling. Shared delivery
coordination should consume normalized receiver identities and deadlines, not
provider protocols, credentials, or command-registration rules.
`runtime/ownership.ts` owns acquisition, renewal, token fencing, and stop order.
Acquisition uses insert or expired-token CAS; background acquisition attempts once.
Foreground configuration can wait up to twenty seconds, checking every 100 ms.
That bounded active-operation wait is separate from idle inventory polling.
The current default live lease lasts fifteen seconds and renews every five
seconds while owned. These renewals are required active transport work and are
not removed in this slice. Existing inbound admission reclaims the exact token
after expiry before accepting buffered callbacks; preserve that safety boundary.
## Proposed wake contract to verify in code
- Reconstruct eligible receiver obligations at startup with bounded UUID pages.
- Notify before supported endpoint/connection eligibility or credential changes
in their existing transactions. Decode scoped hints only at the notification
boundary; narrow normal refreshes to an endpoint or connection.
- Keep locally owned healthy receivers in the inventory cache even when they have
no discovery deadline. Otherwise later removal cannot identify the runtime that
must stop. A cache of only future deadlines is insufficient for this duty.
- A locally healthy unchanged receiver needs no periodic rediscovery query. Its
active lease renewal and socket lifecycle remain owned by the receiver runtime.
- On a concrete foreign lease, schedule the saved expiry as the next takeover
attempt. Reload current authority at that time; a renewed foreign lease yields
its new expiry. This is a configured receiver's takeover obligation, not proof
that a standby with such an obligation can sleep indefinitely.
- Initialization/credential/database failures retain scoped retry/backoff. One
failed receiver must not starve independent companies or a healthy due backlog.
- After ownership loss, finish listener fencing/stop and lease release, then wake
discovery for that endpoint. Do not reinitialize before the old listener stops.
Avoid a callback cycle on shutdown or a permanent retry after disable/removal.
- Preserve a change arriving during refresh, drain, or stop. Use the existing
coordinator's settled hints and memory-only pending predicate; no passive SQL
during drain. Join admitted work before returning from shutdown.
- Retain the explicit/manual reconciliation entry point used by current callers,
or update those callers and their assertions deliberately. Do not leave a
second automatic periodic entry point invoking the old full inventory query.
## Producer audit still required before implementation is accepted
Existing helpers are starting points, not proof of complete coverage:
1. `chat-work-notifications.ts:notifyChatEndpointWork` already covers shared
connection-origin delivery/command wakes. Inspect its callers individually.
2. `chat-channels/configuration.ts`: configure, reconnect, resume, pause, remove,
activation and rollback, including foreground acquisition and ownership waits.
3. `chat-channels/setup-probes.ts`, `provider-effects.ts`, resource reconciliation,
publication failures, and inbound activation: eligibility/health transitions.
4. Generic tool access, removal, health observations, credential/binding changes,
membership removal, and company deletion: these can change a receiver without
calling the endpoint configuration service.
5. `runtime/ownership.ts`: renewal failure, token loss, failed initialization,
runtime invalidation, lease release, and shutdown. A direct callback is suitable
for process-local completion; do not invent a SQL row for that callback.
6. Photon runtime callbacks and configuration: independently verify these paths;
the Discord parent producer audit is not evidence of Photon coverage.
7. Existing live replica/independent-writer assumptions: process-local transaction
hints cannot discover another process's arbitrary configuration writes. Retain
or explicitly integrate the supported cross-process wake contract rather than
silently removing its only recovery path.
## Required evidence
Start with a failing empty-instance regression proving that the aggregate chat
cadence repeatedly queries receiver inventory. Then verify:
- empty, disabled, and retained-history states have no discovery SQL after startup;
- actual first configure/reconnect/disable/remove producers wake or cancel work;
- healthy local ownership adds no inventory scan beyond explicit mutations;
- foreign lease expiry, extension, and takeover preserve single ownership;
- renewal failure triggers bounded recovery only after fencing the old runtime;
- old buffered messages remain fenced after token or credential generation change;
- Photon and Discord preserve their existing distinct transport behavior;
- scoped changes, company deletion, nested transaction commit/rollback, uncertain
commit, page boundaries, failure isolation, concurrent changes, restart,
drain/resume, and shutdown retain obligations;
- shared coordinator behavior without this worker remains unchanged;
- the existing command-registration cases retain saved verification deadlines.
Use real PostgreSQL for lease CAS and transaction ownership, and controlled local
runtime/provider fixtures for lifecycle effects. Existing chat integration tests
exercise manual reconciliation, foreign ownership, credential rotations, and
Photon restart. Inspect those cases before designing fixtures. SQL silence must
be measured after both tracked work and scheduler settlement; a null next-wake
value alone can occur during an admitted reload.
## Sleep limits and remaining stack gates
Active receiver sockets and lease renewals remain physical sleep blockers. Cloud
alarm coverage and admission/sleep-safety proof do not change. This migration
removes idle discovery activity, not the receiver's live transport obligation.
The full original recurring-server inventory remains in scope in the stack ledger.
At audit creation, health #17554 has a passing security scan on 9ef6e4f4c8 but
other current-head checks remain running. Company cleanup #17561 is e8febb30eb.
Discord #17557 is d3187f76a4 with fresh checks required after its failure-path and
CI fixture repairs. The older #17490 E2E cancellation failure, final reconciliation
with master, and all fresh review/check gates remain outstanding. No stack PR was
merged and no new merge authority is implied.
Follow-up acceptance snapshot: health #17554 at 9ef6e4f4c8 now has no pending or
failed checks, passing security, and fresh Greptile 5/5. Its three reviewed findings
(fair batching, real producer coverage, inventory amplification) are implemented,
verified, and their threads resolved. This is acceptance of that current head,
not a claim that the full stack is complete or reconciled with moving master.
Company cleanup and Discord still have live current-head checks.
## Implementation and verification update
The aggregate chat lane now calls voice reconciliation only. Receiver discovery
registers with the existing delivery coordinator. The receiver owner loads UUID
pages of 100, keeps owned identities with no discovery deadline, and processes
25 due receivers per fair company batch. Configuration hints refresh only their
endpoint, connection, or company. A foreign owner schedules its saved lease
expiry; a changed lease is reread before another attempt. Failures retain scoped
30-second exponential backoff capped at five minutes. Dirty scopes suppress their
cached deadlines until refresh succeeds; otherwise an old takeover deadline could
cause repeated useless dispatch while the required refresh is in backoff.
The generic configuration notifier replaces the misleading command-only helper
name. Existing producers now notify both relevant owners. Shared delivery code
still receives only a queue topic and scope hint. Provider selection and lease
keys remain in the chat receiver owner. No schema or new per-worker table exists.
Audited producer groups:
| Owner | Receiver-relevant changes and integration |
| --- | --- |
| Configuration and setup probes | Existing connection helper covers configure, rotation, resume, pause, removal and completed setup before transaction writes |
| Credential persistence and removal | Configuration notifier before reference changes, including Photon |
| Inbound activation and resource reconciliation | Endpoint/connection notifier before status and health transitions |
| Publication, interaction and provider-effect failures | Endpoint configuration notifier before attention/failed-health writes |
| Discord Gateway lifecycle | Receiver hints before recovery/failure transitions; metadata-only writes do not create discovery work |
| Photon transport failure and runtime recovery | Failure callback now signals before attention/disable; existing assert-owned recovery uses the shared endpoint helper |
| Generic tool update/removal, manual health and access revocation | Existing chat connection helper, broadened configuration dispatch, and SDK transport filter |
| Company deletion | Company hint precedes endpoint removal in the owning deletion transaction |
| Ownership loss / invalidation | Local completion callback after listener stop and lease release; suppressed during shutdown |
The additional direct-writer audit found publication-resolution, publication
success, provider-effect success, inbound filtering, and Discord root-thread
writes that only update timestamps. Endpoint display/reach preferences do not
change receiver eligibility or the runtime generation. Photon configuration
invalidates the runtime, writes allocation metadata, and proceeds through the
credential/configuration owner; its allocation metadata is not an independent
receiver identity. Slack/GitHub-only setup writers do not create leased receiver
work. Capability projection in Discord reconciliation retains its existing owner.
A real token-loss regression found a concurrent configuration wake could acquire
another lease while the old listener was still stopping. The ownership map now
retains the stopping owner until listener shutdown and lease release complete.
Acquisition joins that owner's existing stop promise. Current-authority checks
still reject stopping owners. The test uses real timing and holds listener stop;
an earlier fake-clock observation returned before async SQL completed and did not
prove absence of overlap. The strengthened test failed before the ownership fix.
Current passing evidence: 32 receiver/startup cases, 90 selected Discord
integration cases, all 20 Photon channel cases, and server typechecking. The
Discord selection includes configuration, credential rotation, pause/resume,
company deletion, token-loss stop ordering, and existing takeover/admission fences.
The dedicated real-DB worker takeover test also passes, as do 75 command/delivery/
standby consumer cases and full workspace typechecking. This is 218 unique selected
passing cases. Workspace build passes. Remote review/check gates remain pending.
The full local test matrix is not claimed: embedded PostgreSQL is unavailable
on this host, so these database suites use validated disposable external URLs.
Remote CI must run the full matrix before acceptance.
Recorded failures remain part of acceptance: initial receiver teardown omitted
agent deletion; initial Discord pause used incomplete onboarding; failed-scope
backoff initially selected the blocked cached takeover deadline; the real-timing
stop race reproduced overlapping listeners. The first two were fixture defects;
the latter two required implementation repairs. An initial backoff test expected
zero idle holds during failed intent recovery; the correct contract retains the
intent hold until reconciliation succeeds, so the test now checks the retry
deadline and SQL silence rather than weakening that hold.
## Writer and sleep boundaries
Notifications cover the active database owner and its transaction-sharing clients.
Warm standby does not run discovery; activation reconstructs its pending startup
inventory. An already known foreign receiver remains recoverable from lease expiry,
with real token CAS and callback fences. Arbitrary new configuration written by
an independent process or direct SQL is not discovered by process-local hints;
that deployment needs an explicit wake integration. This follows the existing
single active writer contract and must not be described as multi-writer support.
The manual reconciliation method remains available for explicit callers.
Active sockets and lease renewals remain required work, and no Cloud admission
exemption is added. Voice, Teams, publication maintenance, milestones and other
inventory reasons remain separate PRs. The full stack is not complete.
Master was fetched again at 0d73505098. Company prerequisite #17561 at e8febb30eb
and Discord parent #17557 at d3187f76a4 now have no failed or pending checks;
Paperclip Review is skipped. Fresh review-content reconciliation and final stack
rebasing/verification remain distinct from these check results.
## Receiver review repair
The real company-removal producer regression reproduced two inventory reads for
one observed removal. Discovery-initiated stops now suppress their own completion
wake; independent token loss still wakes discovery after listener retirement.
A full reconstruction also covers its already-pending scoped hints. Hints that
arrive during the load remain dirty, and narrower writes cannot bypass a failed
full reconstruction's retry deadline.
Both redundant-load regressions failed before repair. All 45 selected receiver,
startup and real Discord producer tests pass after repair, including the added
full-refresh failure/backoff case. Server typecheck and build pass. Prior broad
consumer/workspace verification remains recorded above; these 45 overlap that
coverage and are not an additional unique total. Publication requires fresh
remote CI and review. The full migration inventory remains outstanding.
## Healthy receiver progress after a slow failed batch
The receiver worker already stores a retry deadline for each failed receiver and
scope. It now supplies that deadline calculation through the coordinator's
`nextRunAtOnError` contract as well. A completed batch can take longer than the
one-second batching interval. When that batch contained an error, the coordinator
previously ignored an already-due healthy backlog and waited for its generic
30-second retry. The changed registration lets healthy due entries proceed while
the failed receiver retains its own exponential backoff.
The existing 105-receiver PostgreSQL regression now runs with both an immediate
failure and a controlled 1,100 ms elapsed interval in the failed provider call.
The latter failed before the repair: its next wake was 30 seconds away although
healthy receivers were due. Afterward, all 105 are attempted before retrying the
failed receiver, which is attempted once during the healthy drain. Recovery then
settles and two former poll periods add no SQL. The assertion helper advances its
own fake clock; the test no longer adds an extra second on every assertion retry.
All 20 receiver lifecycle cases and the server production typecheck pass on the
owning branch. This changes one coordinator registration and adds no provider
branch, queue, table, or independent timer. Current-head CI and combined-stack
verification remain separate gates.
## Current-parent and review acceptance, October 27
The branch normally integrates Discord parent #17557 at
`cedf2030c49c3650a41b76a4767176fd78170819` through `3e957d1204`. Only the
original 27 receiver-owned paths differ from that parent. Health #17554 is now
merged; company cleanup #17561 and Discord #17557 still precede this slice.
Older stack status above describes its recorded source revision.
The latest old review summaries still mention the delay and redundant-stop
findings, but the actual source already contains their repairs: stop-wake
suppression at `1604a9fcba` and the error-deadline registration at `ab56393acc`.
Both were revalidated on the integrated parent rather than duplicated. A guarded
mutation removing `nextRunAtOnError` fails the 1,100 ms slow-provider next-deadline
assertion. The expected error also remains visible in teardown after that early
assertion. A separate mutation reenabling discovery-initiated stop notification
fails the real company-removal producer assertion with two inventory reads
instead of one. Both source files are restored byte for byte after the checks.
Normal worker and real producer suites pass with the original fixes.
The two newly supported external-database fixtures initially skipped migrations.
Actual worker reconstruction and Photon transport-loss cases failed on separate
empty PostgreSQL 17 databases with missing-table errors. Both fixtures now apply
pending migrations before client use. On those same initially empty databases,
the complete receiver file passes all 20 cases and the complete Photon channel
file passes all 20 cases. Existing embedded setup and safety validation remain.
Current acceptance: five complete receiver/coordinator/batching/startup/GitHub
error-projection suites pass 99 cases. Twelve actual Discord automatic-command,
receiver configure/rotation/pause/resume/removal, ownership-loss/takeover, and Slack
held-transaction cases pass separately; 1,581 other chat integration cases are
not counted as run. With the 20 Photon cases this is 131 distinct targeted cases.
The large chat producer suite uses a separate migrated clone. Server production
typecheck and build pass on the integrated source; the later fixture-only changes
add migration setup and do not change production code. Startup listener-count
warnings and expected controlled provider failures remain in the logs.
Required CI at the older `ab56393acc` head failed server shard 12. GitHub's
check-run annotation says the self-hosted runner lost communication with the
server; its grouped-test step is marked cancelled. The retained log ends without
a failed assertion or final test summary. That annotation does not establish the
underlying host, resource, or network cause, and the interrupted job is not a
passing result. The new parent integration and fixture repairs require a fresh
full CI run; local passes do not replace it. Fresh exact-head Greptile and
Ponytail reviews remain mandatory. No deployed sleep or full local suite is claimed.
## Landed-foundation integration acceptance, October 27
This slice normally integrates Discord parent `547fdf2139`, which contains
master `f8563d2507` and landed company-deletion fixes. The original 27-file
scope is unchanged. All owned source matches its precise original patch applied
to that parent. The shared queue catalog also preserves master's task-drain and
native-checkpoint declarations. An initial whole-file identity assertion rejected
those expected parent additions; the corrected proof checks the actual owned
patch against the integrated parent. No failed verification is counted as a pass.
The complete five worker/shared-consumer suites pass 99 cases. Real Discord and
Slack producers pass 12 cases; the complete Photon suite passes 20. This is 131
distinct targeted cases, with 1,581 chat cases deselected. Receiver and Photon
self-migrate separate initially empty databases. The large chat fixture uses a
separate migrated clone. Server typecheck and build pass. Only documentation
changes follow the passing source. Expected controlled provider failures and
startup signal-listener warnings remain recorded. Fresh exact-head review and
all CI must pass before this child lands after Discord. No deployed sleep or
full local suite result is inferred from these checks.
## Compatibility after Discord squash merge, October 27
Discord #17557 landed at `5bf0142d45`. The normal merge of that master caused
ancestry conflicts in overlapping producer paths. Resolution applies only the
recorded 27-file receiver patch from Discord parent `547fdf2139` to actual landed
master. Every server and package production/test file is byte-identical to the
published `6e88ca0201` source that passes 131 cases, typecheck and build. No source
verification is rerun merely for unrelated upstream UI/eval/documentation changes.
The PR-owned delta passes its whitespace check. This integration records that
proof and requires fresh exact-head review and CI before merging into master.
doc/plans/2026-10-25-configuration-worker-drain-repair.md
+16
View File
@@ -43,3 +43,19 @@ 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.
## Receiver discovery — slice 20 / #17572
The strengthened first-receiver drain case fails with active=1 after commit in
/tmp/sleep20-drain-before.log. Apply the same owner-level pending predicate while
retaining all dirty scopes and lease/retry deadlines. All 19 receiver lifecycle
cases pass in /tmp/sleep20-drain-tests.log, including no SQL and an unchanged idle
snapshot for a minute after settlement, then receiver acquisition on resume.
Production server typecheck passes in /tmp/sleep20-drain-typecheck.log. The fresh
migrated database is paperclip_sleep20_drain_20261025_test.
The three configuration owners now have measured drain activity as well as traffic
checks. Their retained timers remain memory-only while paused; they do not provide
external wake coverage. Fresh remote CI/review and downstream propagation remain
required. Origin/master was refreshed to e335265568; the latest change removes
idle task-run UI polling and does not replace these worker repairs.
server/src/__tests__/app-private-hostname-gate.test.ts
+18 -10
View File
@@ -25,7 +25,7 @@ describe("createChatReconciliationCoordinator", () => {
processFailedGitHubWebhookDeliveries,
processPendingDeliveries,
flushPublications,
reconcileProviderRuntimes: async () => undefined,
reconcileVoiceSessions: async () => undefined,
projectRunMilestones: async () => 0,
processPendingSlackFileUploadReceipts: async () => undefined,
processPendingSlackSessionSyncs: async () => undefined,
@@ -74,6 +74,14 @@ describe("createChatReconciliationCoordinator", () => {
},
);
it("keeps receiver inventory out of periodic chat maintenance while preserving voice", () => {
const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
const maintenance = source.slice(source.indexOf("const chatReconciliation ="), source.indexOf("const unsubscribeChatPublicationSignals"));
expect(maintenance).not.toContain("chatChannels.reconcileProviderRuntimes()");
expect(maintenance).toContain("chatChannels.voice.reconcile()");
expect(source).toContain("chatChannels.registerReceiverWork(deliveryWork,");
});
it("wires GitHub recovery into its independent lane", () => {
const source = readFileSync(new URL("../app.ts", import.meta.url), "utf8");
expect(source).toContain(
@@ -135,7 +143,7 @@ describe("createChatReconciliationCoordinator", () => {
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones,
reconcileProviderRuntimes: async () => undefined,
reconcileVoiceSessions: async () => undefined,
processPendingDeliveries: async () => undefined,
flushPublications,
processPendingSlackFileUploadReceipts: async () => undefined,
@@ -198,7 +206,7 @@ describe("createChatReconciliationCoordinator", () => {
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones,
flushPublications,
reconcileProviderRuntimes: async () => undefined,
reconcileVoiceSessions: async () => undefined,
processPendingDeliveries: async () => undefined,
processPendingSlackFileUploadReceipts: async () => undefined,
processPendingSlackSessionSyncs: async () => undefined,
@@ -238,7 +246,7 @@ describe("createChatReconciliationCoordinator", () => {
const slackStatusReleased = new Promise<void>((resolve) => {
releaseSlackStatus = resolve;
});
const reconcileProviderRuntimes = vi.fn(async () => undefined);
const reconcileVoiceSessions = vi.fn(async () => undefined);
const processPendingDeliveries = vi.fn(async () => deliveryReleased);
const flushPublications = vi.fn(async () => undefined);
const processPendingSlackFileUploadReceipts = vi.fn(async () => undefined);
@@ -248,7 +256,7 @@ describe("createChatReconciliationCoordinator", () => {
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones: async () => 0,
reconcileProviderRuntimes,
reconcileVoiceSessions,
processPendingDeliveries,
flushPublications,
processPendingSlackFileUploadReceipts,
@@ -258,7 +266,7 @@ describe("createChatReconciliationCoordinator", () => {
coordinator.reconcile();
await vi.waitFor(() => {
expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(1);
expect(reconcileVoiceSessions).toHaveBeenCalledTimes(1);
expect(processPendingDeliveries).toHaveBeenCalledTimes(1);
expect(flushPublications).toHaveBeenCalledTimes(1);
expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(1);
@@ -266,7 +274,7 @@ describe("createChatReconciliationCoordinator", () => {
});
coordinator.reconcile();
await vi.waitFor(() => {
expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(2);
expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2);
expect(flushPublications).toHaveBeenCalledTimes(2);
expect(processPendingSlackFileUploadReceipts).toHaveBeenCalledTimes(2);
});
@@ -293,14 +301,14 @@ describe("createChatReconciliationCoordinator", () => {
await receiptReleased;
if (rejects) throw lookupError;
});
const reconcileProviderRuntimes = vi.fn(async () => undefined);
const reconcileVoiceSessions = vi.fn(async () => undefined);
const processPendingDeliveries = vi.fn(async () => undefined);
const flushPublications = vi.fn(async () => undefined);
const processPendingSlackSessionSyncs = vi.fn(async () => undefined);
const onError = vi.fn();
const coordinator = createChatReconciliationCoordinator({
projectRunMilestones: async () => 0,
reconcileProviderRuntimes,
reconcileVoiceSessions,
processPendingDeliveries,
flushPublications,
processPendingSlackFileUploadReceipts,
@@ -318,7 +326,7 @@ describe("createChatReconciliationCoordinator", () => {
});
coordinator.reconcile();
await vi.waitFor(() => {
expect(reconcileProviderRuntimes).toHaveBeenCalledTimes(2);
expect(reconcileVoiceSessions).toHaveBeenCalledTimes(2);
expect(processPendingDeliveries).toHaveBeenCalledTimes(2);
expect(flushPublications).toHaveBeenCalledTimes(2);
expect(processPendingSlackSessionSyncs).toHaveBeenCalledTimes(2);
server/src/__tests__/chat-channels.integration.test.ts
+122 -1
View File
@@ -1,3 +1,4 @@
import { notifyChatEndpointConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js";
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";
@@ -7133,6 +7134,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
async function registrationFixture(
mode: "success" | "unavailable" | "unknown" = "success",
receiverOptions: Partial<Pick<ChatChannelServiceOptions, "discordGatewayLeaseTtlMs" | "discordGatewayLeaseRenewalIntervalMs">> = {},
) {
const fixture = await seedCompany();
const applicationId = uniqueDiscordApplicationId();
@@ -7190,7 +7192,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
return Response.json(command);
},
) as unknown as typeof globalThis.fetch;
const context = createService(new FakeChatSdkRuntime(), providerFetch);
const context = createService(new FakeChatSdkRuntime(), providerFetch, receiverOptions);
const endpoint = await context.service.create(
fixture.companyId,
{
@@ -7243,6 +7245,125 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
};
}
it("drives receiver discovery through real configure, rotation, pause, and resume producers", async () => {
const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 600_000, discordGatewayLeaseRenewalIntervalMs: 300_000 });
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
const settle = async () => vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000);
expect(idleWorkSnapshot().active).toBe(0);
expect(coordinator.nextWakeAt()).toBeNull();
}, { timeout: 5000 });
const quiet = async () => {
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction");
try {
await vi.advanceTimersByTimeAsync(20_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled();
} finally { reads.mockRestore(); writes.mockRestore(); }
};
try {
await f.service.registerReceiverWork(coordinator, () => true).ready;
await settle(); await quiet();
await f.configure(); await settle();
const first = f.runtime.get(f.endpoint.id); expect(first).not.toBeNull();
await quiet();
await f.service.configure(f.endpoint.id, { action: "reconnect", credentials: { ...f.credentials, botToken: "rotated-receiver-token" } }, "owner-user");
await settle(); expect(f.runtime.get(f.endpoint.id)).not.toBe(first); expect(first!.shutdown).toHaveBeenCalledOnce();
// Pause requires completed onboarding; configure/reconnect intentionally
// leave this fixture waiting for its first test conversation.
const configured = await f.service.get(f.endpoint.id);
await db.update(chatEndpoints).set({ status: "active", setup: { ...configured.setup, step: "complete" } })
.where(eq(chatEndpoints.id, f.endpoint.id));
await f.service.configure(f.endpoint.id, { action: "pause" }, "owner-user");
await settle(); expect(f.runtime.get(f.endpoint.id)).toBeNull(); await quiet();
await f.service.configure(f.endpoint.id, { action: "resume" }, "owner-user");
await settle(); expect(f.runtime.get(f.endpoint.id)).not.toBeNull(); await quiet();
await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user");
await settle(); expect(f.runtime.get(f.endpoint.id)).toBeNull(); await quiet();
expect(onError).not.toHaveBeenCalled();
} finally {
await coordinator.stop(); await f.service.shutdown(); vi.useRealTimers();
await retireRegistrationFixture(f.service, f.endpoint.id);
}
});
it("does not rediscover a receiver stopped by a company removal refresh", async () => {
const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 600_000, discordGatewayLeaseRenewalIntervalMs: 300_000 });
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
const onError = vi.fn(), coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
let reads: ReturnType<typeof vi.spyOn> | undefined;
try {
await f.configure(); await f.service.registerReceiverWork(coordinator, () => true).ready;
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); });
const owned = f.runtime.get(f.endpoint.id)!;
reads = vi.spyOn(db, "select");
await db.transaction(async tx => {
await notifyChatCompanyConfiguration(tx, f.fixture.companyId);
await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, f.endpoint.id));
});
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull(); });
expect(owned.shutdown).toHaveBeenCalledOnce();
expect(reads).toHaveBeenCalledOnce(); expect(onError).not.toHaveBeenCalled();
} finally {
reads?.mockRestore(); await coordinator.stop(); vi.useRealTimers(); await f.service.shutdown();
await db.update(chatEndpoints).set({ status: "archived" }).where(eq(chatEndpoints.id, f.endpoint.id));
}
});
it("takes over a persisted foreign receiver lease at its deadline through the worker", async () => {
const f = await registrationFixture();
await f.configure(); await f.service.shutdown();
const token = randomUUID(), expiresAt = new Date(Date.now() + 500);
await db.insert(chatEndpointLeases).values({ companyId: f.fixture.companyId, endpointId: f.endpoint.id,
leaseKey: "discord_gateway_runtime", token, expiresAt });
const standby = createService(new FakeChatSdkRuntime(), f.providerFetch);
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
try {
await standby.service.registerReceiverWork(coordinator, () => true).ready;
expect(standby.runtime.get(f.endpoint.id)).toBeNull();
expect(coordinator.nextWakeAt()).toBeGreaterThanOrEqual(expiresAt.getTime());
await vi.waitFor(() => expect(standby.runtime.get(f.endpoint.id)).not.toBeNull(), { timeout: 5000 });
const [lease] = await db.select().from(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, f.endpoint.id));
expect(lease.token).not.toBe(token);
expect(lease.expiresAt.getTime()).toBeGreaterThan(expiresAt.getTime());
expect(onError).not.toHaveBeenCalled();
} finally {
await coordinator.stop(); await retireRegistrationFixture(standby.service, f.endpoint.id);
}
});
it("restarts a receiver after token loss only after the old listener has stopped", async () => {
const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 60_000, discordGatewayLeaseRenewalIntervalMs: 50 });
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
let release!: () => void;
const stopped = new Promise<void>(resolve => { release = resolve; });
try {
await f.configure();
await f.service.registerReceiverWork(coordinator, () => true).ready;
const old = f.runtime.get(f.endpoint.id)!;
old.shutdown.mockImplementation(async () => { await stopped; });
await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, f.endpoint.id));
await vi.waitFor(() => expect(old.shutdown).toHaveBeenCalledOnce());
await db.transaction(tx => notifyChatEndpointConfiguration(tx, f.endpoint));
await new Promise(resolve => setTimeout(resolve, 1500));
expect(f.runtime.get(f.endpoint.id)).toBeNull();
release();
await vi.waitFor(async () => {
expect(f.runtime.get(f.endpoint.id)).not.toBeNull();
expect(f.runtime.get(f.endpoint.id)).not.toBe(old);
expect(idleWorkSnapshot().active).toBe(0);
expect(coordinator.nextWakeAt()).toBeNull();
});
expect(onError).not.toHaveBeenCalled();
} finally {
release(); await coordinator.stop(); vi.useRealTimers();
await retireRegistrationFixture(f.service, f.endpoint.id);
}
});
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 });
server/src/__tests__/chat-receiver-work.test.ts
+245
View File
@@ -0,0 +1,245 @@
import { randomUUID } from "node:crypto";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { eq, sql } from "drizzle-orm";
import { applyPendingMigrations, agents, chatEndpoints, chatEndpointLeases, companies, createDb, toolApplications, toolConnections } from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { notifyDeliveryWork, DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { chatReceiverSchedule } from "../services/chat-channels/runtime/receiver-schedule.js";
import { registerChatReceiverWork } from "../services/chat-channels/runtime/receiver-work.js";
import { notifyChatEndpointConfiguration, notifyChatConnectionConfiguration, notifyChatCompanyConfiguration } from "../services/chat-channels/configuration-work-notifications.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RECEIVER_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("chat receiver discovery", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>, companyId: string, agentId: string, applicationId: string;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let enabled = true, primary = true;
const errors: unknown[] = [];
const reconcile = vi.fn(async (_id: string): Promise<"owned" | "foreign"> => "owned"), stop = vi.fn(async (_id: string) => {});
beforeAll(async () => {
if (!external) temporary = await startEmbeddedPostgresTestDatabase("chat-receivers-");
if (external) await applyPendingMigrations(external);
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
}, 60_000);
beforeEach(async () => {
companyId = randomUUID(); agentId = randomUUID(); applicationId = randomUUID(); enabled = true; primary = true; errors.length = 0;
reconcile.mockReset().mockResolvedValue("owned"); stop.mockReset().mockResolvedValue(undefined);
await db.insert(companies).values({ id: companyId, name: "Receivers", issuePrefix: `R${companyId.slice(0, 7)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Receiver", adapterType: "codex_local" });
await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Receiver", type: "chat" });
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select 1`)));
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
vi.useRealTimers(); vi.restoreAllMocks();
await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId));
await db.delete(agents).where(eq(agents.companyId, companyId));
await db.delete(companies).where(eq(companies.id, companyId));
expect(errors).toEqual([]);
});
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
async function endpoint(patch: Partial<typeof chatEndpoints.$inferInsert> = {}, writer = db) {
const id = patch.id ?? randomUUID(), connectionId = randomUUID();
return writer.transaction(async tx => {
await notifyChatEndpointConfiguration(tx, { id, provider: patch.provider ?? "discord" });
await tx.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: id, uid: id,
transport: "chat_sdk", connectionPurpose: "channel", enabled: true, status: "active" });
return (await tx.insert(chatEndpoints).values({ id, companyId, connectionId, assignedAgentId: agentId,
publicId: id, provider: "discord", status: "active", ...patch }).returning())[0];
});
}
async function start(service = chatReceiverSchedule(db, { reconcile, stop })) {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => enabled && primary, canReconcile: () => primary, onError: error => errors.push(error) });
const worker = registerChatReceiverWork(coordinator, service, () => enabled && primary);
await worker.ready;
return { worker, service };
}
async function settled(expected: number | null = null) {
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000);
expect(idleWorkSnapshot().active).toBe(0);
expect(coordinator!.nextWakeAt()).toBe(expected);
}, { timeout: 5000 });
}
async function quiet(ms = 120_000) {
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute");
try {
await vi.advanceTimersByTimeAsync(ms);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(execute).not.toHaveBeenCalled();
} finally { reads.mockRestore(); writes.mockRestore(); execute.mockRestore(); }
}
it("does no recurring SQL after empty reconstruction", async () => { await start(); await settled(); await quiet(); });
it("leaves disabled, archived, and webhook-only endpoints quiet", async () => {
await endpoint({ status: "paused" }); await endpoint({ status: "archived" }); await endpoint({ provider: "slack" });
const disabled = await endpoint(); await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, disabled.connectionId));
const failed = await endpoint({ status: "attention" }); await db.update(toolConnections).set({ healthStatus: "failed" }).where(eq(toolConnections.id, failed.connectionId));
await start(); await settled(); await quiet(); expect(reconcile).not.toHaveBeenCalled();
});
it.each(["discord", "imessage-photon"] as const)("wakes the first %s receiver and retains its identity until removal", async provider => {
await start(); await settled(); await quiet();
const row = await endpoint({ provider }); await settled();
expect(reconcile).toHaveBeenCalledWith(row.id); await quiet();
await db.transaction(async tx => { await notifyChatConnectionConfiguration(tx, row.connectionId); await tx.delete(chatEndpoints).where(eq(chatEndpoints.id, row.id)); });
await settled(); expect(stop).toHaveBeenCalledWith(row.id); await quiet();
});
it("waits for outer commit and discards rolled-back configuration", async () => {
await start(); await settled();
await expect(db.transaction(async tx => { await endpoint({}, tx as unknown as typeof db); await vi.advanceTimersByTimeAsync(60_000); expect(reconcile).not.toHaveBeenCalled(); throw new Error("rollback"); })).rejects.toThrow("rollback");
await settled(); await quiet();
let id = "";
await db.transaction(async tx => { id = (await endpoint({}, tx as unknown as typeof db)).id; await vi.advanceTimersByTimeAsync(60_000); expect(reconcile).not.toHaveBeenCalled(); });
await settled(); expect(reconcile).toHaveBeenCalledWith(id);
});
it("preserves the first receiver written during drain without passive SQL", async () => {
await start(); await settled(); enabled = false; const row = await endpoint();
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(1000);
expect(idleWorkSnapshot().active).toBe(0);
});
const drained = idleWorkSnapshot();
await quiet(60_000); expect(idleWorkSnapshot()).toEqual(drained); expect(reconcile).not.toHaveBeenCalled();
enabled = true; await settled(); expect(reconcile).toHaveBeenCalledWith(row.id);
});
it("reconstructs after standby becomes primary without restarting the worker", async () => {
await endpoint(); primary = false; await start(); await quiet(60_000);
primary = true; await settled(); expect(reconcile).toHaveBeenCalledOnce(); await quiet();
});
it("waits for a foreign lease, observes its extension, and takes over after release", async () => {
const row = await endpoint(); let owned = false;
reconcile.mockImplementation(async () => owned ? "owned" : "foreign");
const expiry = Date.now() + 60_000;
await db.insert(chatEndpointLeases).values({ companyId, endpointId: row.id, leaseKey: "discord_gateway_runtime", token: randomUUID(), expiresAt: new Date(expiry) });
await start(); await settled(expiry); await quiet(10_000); expect(reconcile).toHaveBeenCalledOnce();
const renewed = expiry + 60_000;
await db.update(chatEndpointLeases).set({ expiresAt: new Date(renewed) }).where(eq(chatEndpointLeases.endpointId, row.id));
await vi.advanceTimersByTimeAsync(expiry - Date.now()); await settled(renewed); expect(reconcile).toHaveBeenCalledTimes(2);
await db.delete(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, row.id)); owned = true;
await vi.advanceTimersByTimeAsync(renewed - Date.now()); await settled(); expect(reconcile).toHaveBeenCalledTimes(3); await quiet();
});
it("wakes after a local receiver stops and then returns to quiet", async () => {
const row = await endpoint(); const { worker } = await start(); await settled(); await quiet();
worker.changed(row.id); await settled(); expect(reconcile).toHaveBeenCalledTimes(2); await quiet();
});
it("retains a producer arriving during its scoped refresh", async () => {
const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load;
await start(service); await settled(); let inserted = false;
service.load = async (...args) => { const rows = await original(...args); if (!inserted) { inserted = true; await endpoint(); } return rows; };
await endpoint(); await settled(); expect(reconcile).toHaveBeenCalledTimes(2); await quiet();
});
it("cancels all cached owned receivers after a company deletion hint", async () => {
await endpoint(); await endpoint(); await start(); await settled();
await db.transaction(async tx => { await notifyChatCompanyConfiguration(tx, companyId); await tx.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); });
await settled(); expect(stop).toHaveBeenCalledTimes(2); await quiet();
});
it("retains transient load failures and does not repeatedly scan during their backoff", async () => {
const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load;
await start(service); await settled(); let broken = true;
service.load = async (...args) => { if (broken) throw new Error("load unavailable"); return original(...args); };
await endpoint(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); });
await quiet(10_000); broken = false; await settled(); expect(reconcile).toHaveBeenCalledOnce(); errors.length = 0;
});
it("does not schedule blocked takeover attempts before a failed scope refresh can retry", async () => {
const row = await endpoint();
reconcile.mockResolvedValue("foreign");
const expiry = Date.now() + 10_000;
await db.insert(chatEndpointLeases).values({ companyId, endpointId: row.id, leaseKey: "discord_gateway_runtime", token: randomUUID(), expiresAt: new Date(expiry) });
const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load;
await start(service); await settled(expiry);
service.load = async () => { throw new Error("scoped inventory unavailable"); };
await db.transaction(tx => notifyChatConnectionConfiguration(tx, row.connectionId));
await vi.advanceTimersByTimeAsync(1000);
await vi.waitFor(() => expect(errors).toHaveLength(1));
expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now() + 20_000);
await quiet(10_000);
service.load = original; reconcile.mockResolvedValue("owned");
await settled(); errors.length = 0; await quiet();
});
it("coalesces a full refresh with its already pending scoped hints", async () => {
const service = chatReceiverSchedule(db, { reconcile, stop });
await start(service); await settled();
const load = vi.spyOn(service, "load");
await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*");
await endpoint({}, tx as unknown as typeof db);
await endpoint({}, tx as unknown as typeof db);
});
await settled(); expect(reconcile).toHaveBeenCalledTimes(2);
expect(load).toHaveBeenCalledOnce(); expect(load.mock.calls[0][0]).toBeUndefined();
load.mockRestore(); await quiet();
});
it("keeps full-refresh backoff when a narrower producer arrives", async () => {
const service = chatReceiverSchedule(db, { reconcile, stop }), original = service.load;
await start(service); await settled();
const load = vi.spyOn(service, "load").mockRejectedValue(new Error("inventory unavailable"));
await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "*"));
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors).toHaveLength(1); });
const retryAt = coordinator!.nextWakeAt();
const row = await endpoint();
await quiet(10_000);
expect(load).toHaveBeenCalledOnce(); expect(coordinator!.nextWakeAt()).toBe(retryAt);
load.mockImplementation(original);
await settled(); expect(reconcile).toHaveBeenCalledWith(row.id);
expect(load).toHaveBeenCalledTimes(2); expect(load.mock.calls[1][0]).toBeUndefined();
errors.length = 0; load.mockRestore(); await quiet();
});
it("recovers malformed scopes without losing a subsequent valid producer", async () => {
await start(); await settled();
await db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.chatReceivers, "unknown:old"));
await endpoint();
await settled(); expect(reconcile).toHaveBeenCalledOnce(); expect(errors.length).toBeGreaterThan(0); errors.length = 0; await quiet();
});
it.each([0, 1100])("drains paged inventory despite a failed receiver after %i ms", async delay => {
for (let i = 0; i < 105; i++) await endpoint();
const service = chatReceiverSchedule(db, { reconcile, stop });
const reads = vi.spyOn(db, "select"); const inventory = await service.load({ kind: "company", id: companyId });
expect(inventory).toHaveLength(105); expect(reads).toHaveBeenCalledTimes(2); reads.mockRestore();
const failedId = inventory[0].id;
reconcile.mockImplementation(async id => {
if (id === failedId) { vi.setSystemTime(Date.now() + delay); throw new Error("bad credential"); }
return "owned";
});
await start(service);
expect(coordinator!.nextWakeAt()).not.toBeNull();
expect(coordinator!.nextWakeAt()!).toBeLessThanOrEqual(Date.now() + 1000);
await vi.waitFor(() => expect(new Set(reconcile.mock.calls.map(([id]) => id)).size).toBe(105), { timeout: 5000 });
expect(reconcile.mock.calls.filter(([id]) => id === failedId)).toHaveLength(1);
reconcile.mockResolvedValue("owned"); await settled(); errors.length = 0; await quiet();
});
it("includes a second company in the first batch despite a larger earlier backlog", async () => {
for (let i = 1; i <= 30; i++) await endpoint({ id: `00000000-0000-4000-8000-${String(i).padStart(12, "0")}` });
const firstCompany = companyId, firstAgent = agentId, firstApplication = applicationId;
const secondCompany = randomUUID(), secondAgent = randomUUID(), secondApplication = randomUUID();
await db.insert(companies).values({ id: secondCompany, name: "Other receiver company", issuePrefix: `R${secondCompany.slice(0, 7)}` });
await db.insert(agents).values({ id: secondAgent, companyId: secondCompany, name: "Receiver", adapterType: "codex_local" });
await db.insert(toolApplications).values({ id: secondApplication, companyId: secondCompany, name: "Receiver", type: "chat" });
try {
companyId = secondCompany; agentId = secondAgent; applicationId = secondApplication;
const other = await endpoint({ id: "ffffffff-ffff-4fff-8fff-ffffffffffff" });
companyId = firstCompany; agentId = firstAgent; applicationId = firstApplication;
await start();
expect(reconcile.mock.calls.slice(0, 25).map(([id]) => id)).toContain(other.id);
await settled(); expect(reconcile).toHaveBeenCalledTimes(31); await quiet();
} finally {
companyId = firstCompany; agentId = firstAgent; applicationId = firstApplication;
await coordinator?.stop(); coordinator = undefined;
await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, secondCompany));
await db.delete(agents).where(eq(agents.companyId, secondCompany));
await db.delete(companies).where(eq(companies.id, secondCompany));
}
});
it("joins admitted runtime initialization during shutdown", async () => {
await start(); await settled(); let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
reconcile.mockImplementation(async () => { await gate; return "owned"; });
await endpoint(); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(reconcile).toHaveBeenCalledOnce(); });
let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; });
await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false);
release(); await stopping; coordinator = undefined; errors.length = 0;
await quiet(); expect(reconcile).toHaveBeenCalledOnce();
});
});
server/src/__tests__/photon/channel.integration.test.ts
+38 -5
View File
@@ -1,3 +1,5 @@
import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.js";
import { resolveExternalTestDatabaseUrl } from "../helpers/external-test-database.js";
import { Readable } from "node:stream";
import type { StorageService } from "../../storage/types.js";
import { createHash, randomUUID } from "node:crypto";
@@ -21,6 +23,7 @@ import {
} from "@photon-ai/advanced-imessage";
import { and, eq, sql } from "drizzle-orm";
import {
applyPendingMigrations,
createDb,
companies,
agents,
@@ -75,15 +78,18 @@ function idleStream<T>(): TypedEventStream<T> {
);
}
describe("iMessage Photon channel control plane", () => {
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>;
let secrets: string;
const oldKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE;
const services: ChatChannelService[] = [];
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
const companyIds: string[] = [];
beforeAll(async () => {
database = await startEmbeddedPostgresTestDatabase("paperclip-photon-");
db = createDb(database.connectionString);
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PHOTON_TEST_DATABASE_URL");
if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-photon-");
if (external) await applyPendingMigrations(external);
db = createDb(external ?? database!.connectionString);
secrets = await mkdtemp(path.join(tmpdir(), "photon-secrets-"));
process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE = path.join(
secrets,
@@ -91,6 +97,7 @@ describe("iMessage Photon channel control plane", () => {
);
}, 30_000);
afterEach(async () => {
await Promise.all(coordinators.splice(0).map(coordinator => coordinator.stop()));
await Promise.all(services.splice(0).map((service) => service.shutdown()));
for (const id of companyIds.splice(0))
await db
@@ -100,6 +107,7 @@ describe("iMessage Photon channel control plane", () => {
vi.restoreAllMocks();
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await database?.cleanup();
if (secrets) await rm(secrets, { recursive: true, force: true });
if (oldKey === undefined)
@@ -364,8 +372,33 @@ describe("iMessage Photon channel control plane", () => {
},
};
}
it("recovers a Photon transport loss through the receiver worker and cancels fatal recovery", async () => {
const t = await setup();
const onError = vi.fn();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
coordinators.push(coordinator);
await t.service.registerReceiverWork(coordinator, () => true).ready;
const first = t.service.runtime.get(t.endpoint.id);
expect(first).not.toBeNull();
await t.callbacks().onPhotonFailure!(new Error("transport interrupted"));
await vi.waitFor(() => {
const current = t.service.runtime.get(t.endpoint.id);
expect(current).not.toBeNull(); expect(current).not.toBe(first);
}, { timeout: 5000 });
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 5000 });
await t.callbacks().onPhotonFailure!(new PhotonError("credentials", "Rejected credentials"));
await vi.waitFor(() => expect(t.service.runtime.get(t.endpoint.id)).toBeNull(), { timeout: 5000 });
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 5000 });
expect(await db.select().from(toolConnections).where(eq(toolConnections.id, t.endpoint.connectionId)))
.toEqual([expect.objectContaining({ enabled: false })]);
expect(onError).not.toHaveBeenCalled();
});
it("keeps publication authentication failures stopped across receiver reconciliation and restart", async () => {
const t = await setup();
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: () => {} });
coordinators.push(coordinator);
await t.service.registerReceiverWork(coordinator, () => true).ready;
const conversation = await t.start();
await t.qualify();
const callbacks = t.callbacks();
@@ -380,8 +413,8 @@ describe("iMessage Photon channel control plane", () => {
expect.objectContaining({ status: "active", enabled: true, healthStatus: "failed" }),
]);
await expect(callbacks.onPhotonAssertOwned!()).rejects.toThrow();
await t.service.reconcileProviderRuntimes();
expect(t.service.runtime.get(t.endpoint.id)).toBeFalsy();
await vi.waitFor(() => expect(t.service.runtime.get(t.endpoint.id)).toBeFalsy(), { timeout: 5000 });
await coordinator.stop();
const restarted = await t.restart();
expect(restarted.runtime.get(t.endpoint.id)).toBeFalsy();
expect(await restarted.get(t.endpoint.id)).toMatchObject({ status: "attention" });
server/src/app.ts
+5 -7
View File
@@ -321,7 +321,7 @@ export function shouldEnablePrivateHostnameGuard(opts: {
}
type ChatReconciliationLane =
| "provider runtimes"
| "voice sessions"
| "deliveries"
| "GitHub webhook recovery"
| "run milestones"
@@ -335,7 +335,7 @@ type ChatReconciliationLane =
* next publication sweep for every endpoint.
*/
export function createChatReconciliationCoordinator(input: {
reconcileProviderRuntimes: () => Promise<unknown>;
reconcileVoiceSessions: () => Promise<unknown>;
processPendingDeliveries: () => Promise<unknown>;
processFailedGitHubWebhookDeliveries?: () => Promise<unknown>;
projectRunMilestones: () => Promise<number>;
@@ -377,7 +377,7 @@ export function createChatReconciliationCoordinator(input: {
return {
reconcile() {
if (stopped) return;
start("provider runtimes", input.reconcileProviderRuntimes);
start("voice sessions", input.reconcileVoiceSessions);
start("deliveries", input.processPendingDeliveries);
if (input.processFailedGitHubWebhookDeliveries) {
start(
@@ -1298,16 +1298,14 @@ export async function createApp(
}
registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive());
chatChannels.registerReceiverWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
emailChannels.start();
const reconcileChatPublicationMaintenance = async () => {
await chatChannels.processPublicationMaintenance();
};
const chatReconciliation = createChatReconciliationCoordinator({
reconcileProviderRuntimes: async () => {
await chatChannels.reconcileProviderRuntimes();
await chatChannels.voice.reconcile();
},
reconcileVoiceSessions: () => chatChannels.voice.reconcile(),
// Chat delivery, action, and Telegram discovery workers now own these
// duties. Keep the coordinator's other maintenance lanes independent.
processPendingDeliveries: async () => undefined,
server/src/services/chat-channels.ts
+33 -5
View File
@@ -1,4 +1,7 @@
import { notifyChatCommandRegistration } from "./chat-channels/command-work-notifications.js";
import { leasedChatProvider } from "./chat-channels/runtime/receiver-policy.js";
import { chatReceiverSchedule } from "./chat-channels/runtime/receiver-schedule.js";
import { registerChatReceiverWork } from "./chat-channels/runtime/receiver-work.js";
import { notifyChatEndpointConfiguration } from "./chat-channels/configuration-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";
@@ -290,7 +293,6 @@ const ORPHAN_FOLLOW_UP_GRACE_MS = 5_000;
const ORPHAN_FOLLOW_UP_MAX_ATTEMPTS = 12;
const PUBLICATION_ENDPOINT_CONCURRENCY = 4;
function leasedChatProvider(provider: string): boolean { return provider === "discord" || provider === "imessage-photon"; }
const DISCORD_GATEWAY_LEASE_POLL_MS = 100;
const PROVIDER_CREDENTIAL_CHECK_TIMEOUT_MS = 25_000;
@@ -563,7 +565,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
const discordGatewayOwnerships = new Map<string, DiscordGatewayOwnership>();
const discordGatewayStopTasks = new Set<Promise<void>>();
let shuttingDown = false;
const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } });
let receiverWork: ReturnType<typeof registerChatReceiverWork> | undefined;
const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, onStopped: id => receiverWork?.changed(id), get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } });
const { receiveSdkMessages } = createChatSdkMessageReceiver({
endpointRecord, processMessage,
ingressOnly: (provider) => provider === "imessage-photon"
@@ -1117,7 +1120,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
runtimeVersions.delete(endpoint.id);
}
if (discordOwnership) {
await stopDiscordGatewayOwnership(discordOwnership, false);
await stopDiscordGatewayOwnership(discordOwnership, { removeRuntime: false });
}
throw error;
}
@@ -1134,6 +1137,30 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
}
}
const receiverSchedule = chatReceiverSchedule(db, {
stop: async id => {
const ownership = discordGatewayOwnerships.get(id);
// Discovery already observed removal; only independent ownership loss needs another wake.
if (ownership) await stopDiscordGatewayOwnership(ownership, { notifyStopped: false });
},
reconcile: async id => {
const record = await endpointRecord(id);
if (!record) return "owned";
try {
const instance = await runtimeFor(record.endpoint, { requireDiscordOwnership: true });
await reconcileDiscordRuntime(id, instance);
return "owned";
} catch (error) {
if (isDiscordGatewayNotOwnedError(error)) return "foreign";
throw error;
}
},
});
function registerReceiverWork(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
receiverWork = registerChatReceiverWork(coordinator, receiverSchedule, () => !shuttingDown && canRun());
return receiverWork;
}
async function reconcileProviderRuntimes() {
if (shuttingDown) {
return { eligible: 0, local: 0, ownedElsewhere: 0, failed: 0 };
@@ -2211,7 +2238,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
? "Started a conversation by mentioning the agent."
: "Sent an empty message.");
let comment!: Awaited<ReturnType<typeof issuesSvc.addComment>>;
await notifyChatCommandRegistration(taskTx, taskEndpoint);
await notifyChatEndpointConfiguration(taskTx, taskEndpoint);
await taskTx
.update(chatEndpoints)
.set({
@@ -3472,6 +3499,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
return {
registerCommandWork,
registerReceiverWork,
voice,
authorizeFastResponse,
slackRegistration,
server/src/services/chat-channels/command-work-notifications.ts
-18
View File
@@ -1,18 +0,0 @@
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 });
}
server/src/services/chat-channels/configuration-work-notifications.ts
+22
View File
@@ -0,0 +1,22 @@
import { notifyChatReceiver, notifyChatConnectionReceiver, notifyChatCompanyReceivers } from "./runtime/receiver-work-notifications.js";
import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared";
import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js";
export async function notifyChatEndpointConfiguration(transaction: object, endpoint: { id: string; provider: ChatProvider }) {
await notifyChatReceiver(transaction, endpoint);
if (endpoint.provider === "discord") {
await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id });
}
}
export async function notifyChatConnectionConfiguration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") {
await notifyChatConnectionReceiver(transaction, connectionId, transport);
if (transport === "chat_sdk") {
await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId });
}
}
export async function notifyChatCompanyConfiguration(transaction: object, companyId: string) {
await notifyChatCompanyReceivers(transaction, companyId);
await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId });
}
server/src/services/chat-channels/credentials.ts
+3 -3
View File
@@ -1,4 +1,4 @@
import { notifyChatCommandRegistration } from "./command-work-notifications.js";
import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js";
import { prepareSlackInventory } from "./slack/inventory.js";
import { prepareGitHubInventory } from "./github/inventory.js";
import { prepareDiscordInventory } from "./discord/inventory.js";
@@ -352,7 +352,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) {
code: "chat_endpoint_credentials_changed",
});
}
await notifyChatCommandRegistration(tx, endpoint);
await notifyChatEndpointConfiguration(tx, endpoint);
await tx
.delete(companySecretBindings)
.where(
@@ -464,7 +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 notifyChatEndpointConfiguration(tx, endpoint);
await tx
.delete(companySecretBindings)
.where(
server/src/services/chat-channels/discord/gateway-events.ts
+7 -1
View File
@@ -1,3 +1,4 @@
import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js";
import { notifyDiscordCommandRegistration } from "./command-work-notifications.js";
import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence, InternalSetupState } from "../runtime-types.js";
@@ -130,7 +131,10 @@ 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 });
if (current.status === "attention") {
await notifyChatReceiver(tx, current);
await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id });
}
await tx
.update(chatEndpoints)
.set({
@@ -159,6 +163,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents
if (current.status === "attention" && !gatewayOwnedAttention) return;
const status =
current.setup.step === "complete" ? "active" : "verifying";
await notifyChatReceiver(tx, current);
await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id });
if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId);
await tx
@@ -189,6 +194,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents
return;
}
await notifyChatReceiver(tx, current);
await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id });
const detail = discordGatewayFailureDetail(event);
const transientGuildOutage = event.type === "guild_unavailable";
server/src/services/chat-channels/imessage-photon/events.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyChatReceiver } from "../runtime/receiver-work-notifications.js";
import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence } from "../../chat-channels/runtime-types.js";
import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
@@ -104,6 +105,7 @@ export function createPhotonEvents(db: Db, options: PhotonEventsOptions) {
"attention",
]);
if (!endpoint) return;
await notifyChatReceiver(tx, endpoint);
await tx
.update(chatEndpoints)
.set({
server/src/services/chat-channels/interactions/callbacks.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { notifyChatCommandRegistration } from "../command-work-notifications.js";
import { notifyChatEndpointConfiguration } from "../configuration-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";
@@ -418,7 +418,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb
["active"],
);
if (currentEndpoint) {
await notifyChatCommandRegistration(tx, currentEndpoint);
await notifyChatEndpointConfiguration(tx, currentEndpoint);
await tx
.update(chatEndpoints)
.set({
server/src/services/chat-channels/provider-effects.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { notifyChatCommandRegistration } from "./command-work-notifications.js";
import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js";
import { notifyChatConnectionHealth } from "./connection-health-notifications.js";
import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js";
@@ -318,7 +318,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) {
);
}
if (input.disposition.kind === "endpoint_attention" && currentEndpoint) {
await notifyChatCommandRegistration(tx, currentEndpoint);
await notifyChatEndpointConfiguration(tx, currentEndpoint);
await tx
.update(chatEndpoints)
.set({
server/src/services/chat-channels/publications/failures.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { notifyChatCommandRegistration } from "../command-work-notifications.js";
import { notifyChatEndpointConfiguration } from "../configuration-work-notifications.js";
import { notifyChatConnectionHealth } from "../connection-health-notifications.js";
import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js";
@@ -190,7 +190,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur
: null;
if (disposition.kind === "endpoint_attention" && currentEndpoint) {
await notifyChatCommandRegistration(tx, currentEndpoint);
await notifyChatEndpointConfiguration(tx, currentEndpoint);
await tx
.update(chatEndpoints)
.set({
server/src/services/chat-channels/resource-reconciliation.ts
+3 -3
View File
@@ -1,4 +1,4 @@
import { notifyChatCommandRegistration } from "./command-work-notifications.js";
import { notifyChatEndpointConfiguration } from "./configuration-work-notifications.js";
import { notifyChatConnectionHealth } from "./connection-health-notifications.js";
import type { ConnectionManagementContext } from "../connection-management.js";
import { imessageResourceParticipants } from "./imessage-photon/resources.js";
@@ -511,7 +511,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc
: currentEndpoint.status === "attention"
? "verifying"
: currentEndpoint.status;
await notifyChatCommandRegistration(tx, currentEndpoint);
await notifyChatEndpointConfiguration(tx, currentEndpoint);
if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId);
await tx
.update(chatEndpoints)
@@ -541,7 +541,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc
.where(eq(toolConnections.id, currentEndpoint.connectionId));
}
} else {
await notifyChatCommandRegistration(tx, currentEndpoint);
await notifyChatEndpointConfiguration(tx, currentEndpoint);
await tx
.update(chatEndpoints)
.set({
server/src/services/chat-channels/runtime/ownership.ts
+9 -7
View File
@@ -1,3 +1,4 @@
import { receiverLeaseKey as leaseKeyForProvider } from "./receiver-policy.js";
import type { EndpointRow, DbTransaction, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js";
import { randomUUID } from "node:crypto";
import { and, eq, lte } from "drizzle-orm";
@@ -17,12 +18,12 @@ type OwnershipOptions = {
discordGatewayOwnerships: Map<string, DiscordGatewayOwnership>;
discordGatewayStopTasks: Set<Promise<void>>;
isShuttingDown: () => boolean;
onStopped?: (endpointId: string) => void;
readonly discordGatewayLeaseTtlMs?: number;
readonly discordGatewayLeaseRenewalIntervalMs?: number;
readonly discordGatewayLeaseWaitMs?: number;
readonly renewDiscordGatewayLease?: (input: { companyId: string; credentialFingerprint: string; endpointId: string; expiresAt: Date; generation: number; leaseKey: string; token: string }) => Promise<boolean>;
};
const DISCORD_GATEWAY_LEASE_KEY = "discord_gateway_runtime";
const DISCORD_GATEWAY_LEASE_TTL_MS = 15_000;
const DISCORD_GATEWAY_LEASE_WAIT_MS = 20_000;
const DISCORD_GATEWAY_LEASE_POLL_MS = 100;
@@ -73,7 +74,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
async function stopDiscordGatewayOwnership(
ownership: DiscordGatewayOwnership,
removeRuntime = true,
stopOptions: { removeRuntime?: boolean; notifyStopped?: boolean } = {},
): Promise<void> {
if (ownership.stopPromise) return ownership.stopPromise;
ownership.stopping = true;
@@ -81,12 +82,9 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
clearInterval(ownership.renewTimer);
ownership.renewTimer = null;
}
if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) {
discordGatewayOwnerships.delete(ownership.endpointId);
}
ownership.stopPromise = (async () => {
if (
removeRuntime &&
stopOptions.removeRuntime !== false &&
ownership.context.endpointRuntime &&
runtime.get(ownership.endpointId) === ownership.context.endpointRuntime
) {
@@ -104,6 +102,10 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
// The listener is stopped before the lease is released. A standby can
// therefore acquire immediately without overlapping Gateway sessions.
await releaseDiscordGatewayLeaseRow(ownership);
if (discordGatewayOwnerships.get(ownership.endpointId) === ownership) {
discordGatewayOwnerships.delete(ownership.endpointId);
}
if (!options.isShuttingDown() && stopOptions.notifyStopped !== false) options.onStopped?.(ownership.endpointId);
})();
discordGatewayStopTasks.add(ownership.stopPromise);
void ownership.stopPromise.finally(() => {
@@ -304,7 +306,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
context: RuntimeContext,
waitForOwnership: boolean,
): Promise<DiscordGatewayOwnership | null> {
const receiverLeaseKey = endpoint.provider === "imessage-photon" ? "photon_receiver_runtime" : DISCORD_GATEWAY_LEASE_KEY;
const receiverLeaseKey = leaseKeyForProvider(endpoint.provider)!;
const local = discordGatewayOwnerships.get(endpoint.id);
if (
local &&
server/src/services/chat-channels/runtime/receiver-policy.ts
+13
View File
@@ -0,0 +1,13 @@
import type { ChatProvider } from "@paperclipai/shared";
export const RECEIVER_LEASE_KEYS = {
discord: "discord_gateway_runtime",
"imessage-photon": "photon_receiver_runtime",
} as const satisfies Partial<Record<ChatProvider, string>>;
export function receiverLeaseKey(provider: string) {
return Object.hasOwn(RECEIVER_LEASE_KEYS, provider)
? RECEIVER_LEASE_KEYS[provider as keyof typeof RECEIVER_LEASE_KEYS] : undefined;
}
export const leasedChatProvider = (provider: string) => receiverLeaseKey(provider) !== undefined;
server/src/services/chat-channels/runtime/receiver-schedule.ts
+53
View File
@@ -0,0 +1,53 @@
import { and, eq, gt, inArray, ne, or } from "drizzle-orm";
import { chatEndpoints, chatEndpointLeases, toolConnections, type Db } from "@paperclipai/db";
import type { ChatProvider } from "@paperclipai/shared";
import { RECEIVER_LEASE_KEYS, receiverLeaseKey } from "./receiver-policy.js";
import type { ReceiverScope } from "./receiver-work-notifications.js";
export type ReceiverDeadline = { id: string; companyId: string; connectionId: string; dueAt: number | null };
export function chatReceiverSchedule(db: Db, receiver: {
reconcile: (id: string) => Promise<"owned" | "foreign">;
stop: (id: string) => Promise<void>;
}) {
async function load(scope?: ReceiverScope, signal?: AbortSignal): Promise<ReceiverDeadline[]> {
const result: ReceiverDeadline[] = [];
let cursor: string | undefined;
while (true) {
signal?.throwIfAborted();
const rows = await db.select({ id: chatEndpoints.id, companyId: chatEndpoints.companyId, connectionId: chatEndpoints.connectionId })
.from(chatEndpoints).innerJoin(toolConnections, and(eq(toolConnections.id, chatEndpoints.connectionId), eq(toolConnections.companyId, chatEndpoints.companyId)))
.where(and(
inArray(chatEndpoints.provider, Object.keys(RECEIVER_LEASE_KEYS) as ChatProvider[]),
inArray(chatEndpoints.status, ["verifying", "active", "attention"]),
eq(toolConnections.status, "active"), eq(toolConnections.enabled, true),
or(ne(chatEndpoints.status, "attention"), ne(toolConnections.healthStatus, "failed")),
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);
result.push(...rows.map(row => ({ ...row, dueAt: 0 })));
if (rows.length < 100) return result;
cursor = rows[rows.length - 1].id;
}
}
async function check(id: string, signal?: AbortSignal): Promise<ReceiverDeadline | null> {
const scope = { kind: "endpoint", id } as const;
const [before] = await load(scope, signal);
if (!before) { await receiver.stop(id); return null; }
signal?.throwIfAborted();
const ownership = await receiver.reconcile(id);
const [after] = await load(scope, signal);
if (!after) { await receiver.stop(id); return null; }
if (ownership === "owned") return { ...after, dueAt: null };
const [endpoint] = await db.select({ provider: chatEndpoints.provider }).from(chatEndpoints).where(eq(chatEndpoints.id, id));
if (!endpoint) { await receiver.stop(id); return null; }
const key = receiverLeaseKey(endpoint.provider);
const [lease] = key ? await db.select({ expiresAt: chatEndpointLeases.expiresAt }).from(chatEndpointLeases).where(and(
eq(chatEndpointLeases.companyId, after.companyId), eq(chatEndpointLeases.endpointId, id), eq(chatEndpointLeases.leaseKey, key),
)).limit(1) : [];
return { ...after, dueAt: Math.max(Date.now() + 1000, lease?.expiresAt.getTime() ?? 0) };
}
return { load, check, stop: receiver.stop };
}
server/src/services/chat-channels/runtime/receiver-work-notifications.ts
+24
View File
@@ -0,0 +1,24 @@
import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js";
import { leasedChatProvider } from "./receiver-policy.js";
export type ReceiverScope = { kind: "endpoint" | "connection" | "company"; id: string };
export function receiverScope(hint: string): ReceiverScope | undefined {
if (hint === "*") return undefined;
const [kind, id, extra] = hint.split(":");
if (!id || extra !== undefined || (kind !== "endpoint" && kind !== "connection" && kind !== "company"))
throw new Error("Invalid receiver discovery scope");
return { kind, id };
}
export const receiverScopeHint = (scope: ReceiverScope) => `${scope.kind}:${scope.id}`;
export async function notifyChatReceiver(transaction: object, endpoint: { id: string; provider: ChatProvider }) {
if (leasedChatProvider(endpoint.provider))
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "endpoint", id: endpoint.id }));
}
export async function notifyChatConnectionReceiver(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") {
if (transport === "chat_sdk")
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "connection", id: connectionId }));
}
export async function notifyChatCompanyReceivers(transaction: object, companyId: string) {
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatReceivers, receiverScopeHint({ kind: "company", id: companyId }));
}
server/src/services/chat-channels/runtime/receiver-work.ts
+91
View File
@@ -0,0 +1,91 @@
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 { ReceiverDeadline, chatReceiverSchedule } from "./receiver-schedule.js";
import { receiverScope, receiverScopeHint, type ReceiverScope } from "./receiver-work-notifications.js";
function matches(entry: ReceiverDeadline, scope?: ReceiverScope) {
return !scope || (scope.kind === "endpoint" ? entry.id : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id;
}
export function registerChatReceiverWork(
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
service: ReturnType<typeof chatReceiverSchedule>,
canRun: () => boolean,
) {
const dirty = new Set<string>(["*"]);
const entries = new Map<string, ReceiverDeadline>();
const failures = new Map<string, { count: number; retryAt: number }>();
const scopeFailures = new Map<string, { count: number; retryAt: number }>();
let earliest = 0;
let lastCompany: string | null = null;
function failed(map: typeof failures, id: string) {
const count = Math.min((map.get(id)?.count ?? 0) + 1, 5);
map.set(id, { count, retryAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) });
}
function dirtyScopes() {
return [...dirty].map(hint => { try { return receiverScope(hint); } catch { return undefined; } });
}
function next() {
let nextAt: number | null = null;
const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); };
for (const hint of dirty.has("*") ? ["*"] : dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0));
const unavailable = dirtyScopes();
for (const entry of entries.values()) if (entry.dueAt !== null && !unavailable.some(scope => matches(entry, scope)))
take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0));
return nextAt;
}
const worker = coordinator.register(DELIVERY_QUEUES.chatReceivers, {
retryMs: 30_000,
pausePassiveMaintenance: true,
onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); },
run: async signal => {
if (!canRun() || Date.now() < earliest) return;
earliest = Date.now() + 1000;
const errors: unknown[] = [];
const requested = (dirty.has("*") ? ["*"] : [...dirty]).filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
const covered = requested.includes("*") ? [...dirty] : requested;
for (const hint of covered) dirty.delete(hint);
for (let index = 0; index < requested.length; index++) {
if (signal.aborted || !canRun()) { for (const hint of requested.includes("*") ? covered : requested.slice(index)) dirty.add(hint); return; }
const hint = requested[index];
let scope: ReceiverScope | undefined;
let decoded = false;
try {
scope = receiverScope(hint); decoded = true;
const rows = await service.load(scope, signal), retained = new Set(rows.map(row => row.id));
for (const entry of entries.values()) if (matches(entry, scope) && !retained.has(entry.id)) {
await service.stop(entry.id);
entries.delete(entry.id); failures.delete(entry.id);
}
for (const entry of rows) entries.set(entry.id, entry);
if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);
} catch (error) {
const retryHint = decoded ? hint : "*";
dirty.add(retryHint); failed(scopeFailures, retryHint); errors.push(error);
}
}
const unavailable = dirtyScopes();
const due = [...entries.values()].filter(entry => entry.dueAt !== null && !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) entries.set(entry.id, current); else entries.delete(entry.id);
failures.delete(entry.id);
} catch (error) { failed(failures, entry.id); errors.push(error); }
}
if (errors.length) throw new AggregateError(errors, "Chat receiver discovery 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,
nextRunAtOnError: next,
});
return { ready: worker.ready, changed(id: string) {
dirty.add(receiverScopeHint({ kind: "endpoint", id })); worker.wake();
} };
}
server/src/services/chat-work-notifications.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { notifyChatConnectionCommandRegistration } from "./chat-channels/command-work-notifications.js";
import { notifyChatConnectionConfiguration } from "./chat-channels/configuration-work-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
export async function notifyChatDeliveryWork(tx: object): Promise<void> {
@@ -38,5 +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);
await notifyChatConnectionConfiguration(tx, connectionId);
}
server/src/services/company-deletion.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { notifyChatCompanyCommandRegistration } from "./chat-channels/command-work-notifications.js";
import { notifyChatCompanyConfiguration } from "./chat-channels/configuration-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";
@@ -65,7 +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 notifyChatCompanyConfiguration(tx, id);
await chatCompanyDeletion.deleteCompanyData(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id);
server/src/services/delivery-work-notifications.ts
+1
View File
@@ -43,6 +43,7 @@ export const DELIVERY_QUEUES = {
chatGitHubPublications: "chat-github-publications",
chatFailedRunRetries: "chat-failed-run-retries",
chatTelegramMaintenance: "chat-telegram-maintenance",
chatReceivers: "chat-receivers",
chatCommandRegistration: "chat-command-registration",
chatTelegramSubscriptions: "chat-telegram-subscriptions",
chatSessionSyncs: "chat-session-syncs",
server/src/services/tool-access/catalog/health.ts
+2 -2
View File
@@ -1,4 +1,4 @@
import { notifyChatConnectionCommandRegistration } from "../../chat-channels/command-work-notifications.js";
import { notifyChatConnectionConfiguration } from "../../chat-channels/configuration-work-notifications.js";
import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js";
import { connectionActorUserId } from "../actor.js";
import { assertSupportedConnection } from "../connections/setup-policy.js";
@@ -50,7 +50,7 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d
) {
const now = new Date();
const [updated] = await writeToolConnectionHealthChange(db, connection, async tx => {
await notifyChatConnectionCommandRegistration(tx, connection.id, connection.transport);
await notifyChatConnectionConfiguration(tx, connection.id, connection.transport);
return tx.update(toolConnections)
.set({
healthStatus: status,