diff --git a/doc/idle-discord-command-registration.md b/doc/idle-discord-command-registration.md new file mode 100644 index 00000000000..1016ffa794b --- /dev/null +++ b/doc/idle-discord-command-registration.md @@ -0,0 +1,100 @@ +# Discord command registration while idle + +Command registration uses the shared delivery coordinator and existing saved +registration deadlines. It no longer runs as a side effect of the recurring +Gateway runtime reconciliation. Foreground configure and resume still attempt +registration immediately under their existing credential lease. + +## Existing obligations and authority + +The registration store owns `chat_actions` records of kind +`discord_command_registration`. The existing `chat_discord_command_owners` table +retains the application namespace owner even after endpoint or company deletion. +No table, trigger, listener connection, or private timer is added. + +A valid successful receipt has a five-minute remote verification deadline. Other +saved outcomes retain their existing retryAt, including Retry-After. An indefinite +retry remains parked. A genuinely unowned namespace is immediately due. Invalid +receipts, foreign owners, and orphaned ownership tombstones do not become new +registration work. The deadline reader is only a hint; the existing protocol still +checks authority, generation, credentials, identity, and receipt state before +provider operations and commits. An uncertain create is recovered by reading +remote state, never by blindly repeating a create. + +Eligible endpoints are Discord, verifying/active/attention, with an active enabled +connection and a known bot/application ID and guild ID. Attention plus failed +connection health remains excluded. These are the old runtime discovery filters; +company status and unrelated endpoint metadata do not add new policy. + +## Scheduling and lifecycle + +Startup reconstructs eligible state in UUID keyset pages of 100. Committed hints +refresh only the affected endpoint, connection, or deleted company. Refresh and +retry passes have a 30-second minimum spacing. There is no periodic inventory +scan after all hints settle. Due dispatch rechecks persisted eligibility and the +saved deadline, invokes the command owner, and reloads its durable result. + +Each pass admits at most 25 commands, taking turns across companies and preserving +oldest-first order within each company. A company cursor rotates between passes, +including when more companies are due than fit in one pass. A failed or no-progress +attempt backs off from 30 seconds to five minutes; a future stored deadline wins. +A returned promise alone does not prove registration settled. + +Dirty scopes received during work remain pending. Drain and warm standby retain +requests without passive SQL. Discovery checks cancellation between rows/pages. +Shutdown stops the coordinator before the chat service and joins an already +admitted provider operation through settlement. A local Gateway is refreshed +through its existing owner after command work; command failure does not replace +or tear down a healthy listener. + +## Producer coverage + +| Producer | Wake contract | +| --- | --- | +| Registration store | Register intent before initial owner/action insert, state/receipt transitions, and retry-result updates in their existing transaction | +| Configure, reconnect, resume, pause, remove | Existing endpoint-work helper now also signals command scheduling by connection ID | +| Setup test completion | The same endpoint-work helper covers activation | +| Credential reference publication/removal | Signal before binding and reference writes, while retaining the credential lease | +| Gateway connecting, ready, availability and failure callbacks | Scoped endpoint signal when health/status can change eligibility; existing runtime generation fences remain | +| Provider availability reconciliation | Signal available/restored and disabled/removed endpoints | +| Provider-effect, publication and interaction failures | Signal before endpoint attention and failed-health writes | +| Accepted inbound task mutation | Signal the endpoint before its status/connection activation writes; message metadata-only writers stay exempt | +| Generic tool connection API | Existing chat connection transaction helper covers enable/disable and credential/configuration changes | +| Generic connection removal | Signal the SDK connection before archival | +| Generic manual health observation | Chat-owned dispatcher signals SDK connections; other transports do not wake this worker | +| Membership/credential revocation | SDK connections signal before reference removal or disablement | +| Company deletion | Signal the company before the company cascade; namespace tombstones survive | + +Draft endpoint creation has no eligible identity and waits for configuration. +Endpoint display names, reach preferences, resource metadata, publication/event +timestamps, and capability-only runtime projection do not create a registration +obligation. Other providers' setup, OAuth, and registration protocols do not +become Discord work. Direct SQL and independent writers still require explicit +wake integration; process-local hints are not cross-replica notification delivery. + +## Verification and limits + +Selected local verification covers 38 real-PostgreSQL protocol/lifecycle cases, +three fair-batch cases, 91 Discord integration cases, 487 tool-access/access +consumers, and 20 chat-worker/startup consumers: 639 unique tests. Workspace +typechecking and build pass; the final small fairness addition also receives +server type/build verification before publication. The full local `pnpm test:run` +is not claimed: embedded PostgreSQL is unavailable on this host. Remote CI must +complete the full matrix before acceptance. + +The large-backlog fairness regression failed before selection was repaired. +Other observed failures were fixture issues: duplicate live-bot IDs, uncertain +transactions retained by prior fault tests, installing a fake clock after creating +a real timer, pausing a connection still in setup, and removing a fixture twice. +Repeated broad runs on a reused database also found an unchanged hard-coded bot +ID and a global comment-body assertion reading earlier fixtures. The final broad +Discord run uses a fresh schema-only disposable database and passes all 91 cases; +production constraints and the unrelated assertions were not relaxed. See the +[dated audit](plans/2026-10-25-discord-command-registration-sleep-audit.md). + +This migration proves the command worker's quiet periods. Runtime inventory, +receipt-to-runtime projection, active Gateway lease renewal, persistent sockets, +voice, Teams, and other chat maintenance remain separate obligations. The remaining +runtime poll can still read command receipts; it no longer initiates command +registration or remote verification. Cloud sleep eligibility and durable alarm +coverage are unchanged. A timer in this process cannot wake a stopped host. diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index b8bac4cd919..91b3a999a29 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1052,3 +1052,20 @@ Tool-health #17554 at `da4089960619d3da2112a4eed3cc7a3d97dbebf8` has passing Actions checks and fresh Greptile 5/5 after the failed-probe warning repair. Paperclip Review remains queued, so final acceptance is still open. Its PR body now records 771 selected local tests and the outstanding gate. No PR was merged. + + +## Slice 19 integration + +Discord registration now has complete producer wiring and a startup worker; +recurring runtime reconciliation no longer invokes its command protocol. The +[contract](../idle-discord-command-registration.md) records retained verification +policy, scoped producer coverage, supported writers and separate runtime duties. +Local selected coverage is 639 unique passing tests, with workspace types/build +passing before the final fair-batch addition. Final server typechecking and build +also pass. Remote publication/gates remain open. + +The latest health Paperclip Review completed neutral, with a 4/5 assessment and +follow-ups for cross-company backlog fairness and direct service-producer tests. +These remain owning-PR work despite green Actions and Greptile 5/5. The same +fairness defect was reproduced and repaired in the Discord worker. Shared helper +commit `e77979d4fc` is ready to carry into #17554. No PR is newly accepted or merged. diff --git a/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md b/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md index 33e94a3d39a..eeb66f81b1d 100644 --- a/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md +++ b/doc/plans/2026-10-25-discord-command-registration-sleep-audit.md @@ -2,13 +2,13 @@ Parent: tool-health PR #17554, head `da4089960619d3da2112a4eed3cc7a3d97dbebf8`. The parent has 771 selected passing local tests, passing workspace typecheck and -build, passing Actions checks, and fresh Greptile 5/5. Paperclip Review is queued; -it is not accepted or merged. This branch owns +build, passing Actions checks, and fresh Greptile 5/5. Paperclip Review completed neutral with a 4/5 assessment and follow-ups for +company fairness and real-producer coverage. It is not accepted or merged. This branch owns one newly separated reason from the existing chat-runtime inventory. ## Evidence and boundary -`app.ts` calls `reconcileProviderRuntimes` from its recurring chat maintenance +At the parent revision, `app.ts` calls `reconcileProviderRuntimes` from its recurring chat maintenance lane. That service discovers eligible Discord and Photon endpoints, restores runtime ownership, and calls `discord/runtime-reconciliation.ts`. The latter calls `reconcileDiscordCommands` before inspecting registered command state and @@ -65,9 +65,9 @@ Persistent transports remain host-sleep limitations even after discovery is quie `app.ts` also invokes `chatChannels.voice.reconcile()` and Teams maintenance in this lane. Voice, Teams, milestones, wake notices, terminal confirmations, interaction continuations, and GitHub history recovery remain distinct duties. -The scheduling foundation is implemented below, but is not wired into application -startup. The existing runtime backstop remains until the producer audit and -integration tests are complete. No slice-19 PR is published yet. +The worker is now wired into application startup, and runtime reconciliation no +longer initiates registration. Runtime discovery and receipt projection remain +separate. Publication and remote acceptance gates are still pending. ## Implemented scheduling foundation, October 25 @@ -113,7 +113,7 @@ the scoped worker. That preserves the requirement that real uncertain writes must be reconciled before an instance can become quiet. Final evidence is in `/tmp/sleep19-cancellation-tests.log` and `/tmp/sleep19-final-types.log` on this host. -## Required next integration +## Integration checklist at the foundation checkpoint - Wire configuration, readiness, provider failures, generic connection mutation, connection removal, and company deletion producers to the scoped topic. Audit @@ -128,3 +128,49 @@ must be reconciled before an instance can become quiet. Final evidence is in reviews and final reconciliation with master remain mandatory. - Continue every remaining original inventory item and unresolved earlier PR gate. This foundation does not finish this slice or the full goal. + + +## Integrated owner and verification + +The production wiring and producer audit are complete for this slice; see +[the behavior and producer table](../idle-discord-command-registration.md). +The shared application wiring only registers the chat-owned command worker. +Discord scheduling policy stays under `chat-channels/discord/`; shared health, +access and deletion callers use neutral chat notification contracts. The existing +runtime reconciler no longer calls the command protocol. Direct configure/resume +calls remain unchanged, and the worker refreshes an existing local runtime after +command settlement through its current owner. + +The actual configure/removal integration test proves a first configuration after +empty startup publishes its deadline, removal cancels it, and the removed worker +has no further reads or transactions. The real-protocol suite proves a committed +pause arriving during provider I/O is retained and eventually disarms the worker. +The parent health review identified company backlog starvation, and an equivalent +Discord worker regression failed before repair. Shared fair-batch selection now +gives each due company a turn, with a rotating cursor across bounded passes. +The isolated fairness helper commit is `e77979d4fc` for application to the parent. + +Local final evidence: +- `/tmp/sleep19-fairness-after.log`: 41 tests pass (38 protocol/worker + 3 selection). +- `/tmp/sleep19-discord-isolated.log`: 91 selected Discord integrations pass; + 1,327 deselected cases are not counted as passes. +- `/tmp/sleep19-connection-consumers.log`: 487 tool-access/access tests pass. +- `/tmp/sleep19-startup-consumers.log`: 20 chat coordinator/startup tests pass. +- `/tmp/sleep19-workspace-types.log` and `/tmp/sleep19-workspace-build.log`: + workspace typechecking and build pass before the final fairness addition. +- `/tmp/sleep19-fair-types.log` and `/tmp/sleep19-final-server-build.log`: final + server typechecking and build pass after the fairness addition. + +The initial integrated configuration test used a real timer before installing its +fake clock, then attempted pause while the endpoint was still verifying, and then +removed its fixture twice. Those setup/cleanup mistakes were repaired; no product +assertion was weakened. Repeated broad runs reused fixture data: inspection found +one paused endpoint for the unchanged hard-coded concurrent-identity test and two +retained contender comments for the unchanged global-body assertion. A fresh +schema-only database resolves that cross-run contamination. Dynamic registration +fixture IDs are now unique across processes, while ownership tombstones remain. + +Next gates: publish the stacked PR, complete all remote checks +and reviews, repair the health parent's fairness and producer-coverage findings, +and propagate its changes. Full original inventory and earlier unresolved CI +remain required. No merge is authorized. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index c2ef1b9a34d..c2014ea8a80 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -782,11 +782,13 @@ class FakeChatSdkRuntime { const TEST_SLACK_BOT_SCOPES = "app_mentions:read,assistant:write,channels:history,channels:read,chat:write,commands,files:read,files:write,groups:history,groups:read,im:history,im:read,mpim:history,mpim:read,reactions:read,reactions:write,users:read"; +const discordApplicationPrefix = 1_400_000_000_000_000_000n + + BigInt(`0x${randomUUID().replaceAll("-", "").slice(0, 12)}`) * 1000n; let discordApplicationSequence = 0n; function uniqueDiscordApplicationId() { discordApplicationSequence += 1n; - return (123_456_789_012_345_678n + discordApplicationSequence).toString(); + return (discordApplicationPrefix + discordApplicationSequence).toString(); } function fakeSlackFetch(botId = `U-BOT-${randomUUID()}`) { @@ -6600,6 +6602,17 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...context, endpoint, callbacks }; } + async function recoverDiscordCommands(service: ChatChannelService) { + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await service.registerCommandWork(coordinator, () => true).ready; + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); + } + } + describe("Discord automatic command registration", () => { async function retireRegistrationFixture( service: ChatChannelService, @@ -6610,7 +6623,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .select() .from(chatDiscordCommandOwners) .where(eq(chatDiscordCommandOwners.endpointId, endpointId)); - await service.configure(endpointId, { action: "remove" }, "owner-user"); + if ((await service.get(endpointId))?.status !== "archived") { + await service.configure(endpointId, { action: "remove" }, "owner-user"); + } expect((await service.get(endpointId))?.status).toBe("archived"); // Removal retires only this fixture's runtime; immutable application // ownership stays available to the no-adoption/tombstone checks. @@ -6737,6 +6752,35 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; } + it("learns the first configured command and cancels its deadline when the endpoint is removed", async () => { + const f = await registrationFixture(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await f.service.registerCommandWork(coordinator, () => true).ready; + expect(coordinator.nextWakeAt()).toBeNull(); + await f.configure(); + expect(f.calls).toEqual(["GET", "POST"]); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now())); + const deadline = coordinator.nextWakeAt(); + expect(deadline).toBeGreaterThan(Date.now() + 200_000); + await f.service.configure(f.endpoint.id, { action: "remove" }, "owner-user"); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull()); + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(120_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + expect(f.calls).toEqual(["GET", "POST"]); + expect(onError).not.toHaveBeenCalled(); + } finally { + await coordinator.stop(); vi.useRealTimers(); + await retireRegistrationFixture(f.service, f.endpoint.id); + } + }); + it("keeps mention setup usable on registration failure and upgrades automatically when due", async () => { const f = await registrationFixture("unavailable"); try { @@ -6750,7 +6794,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ).toBeTypeOf("function"); f.available(); await f.makeDue(); - await f.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(f.service); expect( (await f.service.get(f.endpoint.id))?.capabilities.slashCommands, ).toBe(true); @@ -6778,6 +6822,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await f.makeDue(); restarted = createService(new FakeChatSdkRuntime(), f.providerFetch); await restarted.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(restarted.service); expect( (await restarted.service.get(f.endpoint.id))?.capabilities .slashCommands, @@ -6852,7 +6897,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }; f.available(false); await f.makeDue(); - await f.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(f.service); expect(f.runtime.endpoints.get(f.endpoint.id)).toBe(original); expect(f.runtime.configurations.get(f.endpoint.id)?.callbacks).toBe( originalCallbacks, @@ -6893,7 +6938,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ), ); await f.makeDue(); - await f.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(f.service); expect( (await f.service.get(f.endpoint.id))?.capabilities.slashCommands, ).toBe(false); @@ -51278,6 +51323,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { }); coldContexts.push(cold); await cold.service.reconcileProviderRuntimes(); + await recoverDiscordCommands(cold.service); return { ...cold, adapter: await parser( diff --git a/server/src/app.ts b/server/src/app.ts index 2205bd47203..ac1666f56d2 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1298,6 +1298,7 @@ export async function createApp( } registerChatDeliveryWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); registerChatActionWork(deliveryWork, chatChannels, () => !isIdleTaskDrainActive()); + chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); emailChannels.start(); const reconcileChatPublicationMaintenance = async () => { await chatChannels.processPublicationMaintenance(); diff --git a/server/src/services/access.ts b/server/src/services/access.ts index d80bde25f99..279a0753e22 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointWork } from "./chat-work-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -130,6 +131,7 @@ export function accessService(db: Db) { }> = []; let connectionRefs: Array<{ id: string; + transport: typeof toolConnections.$inferSelect.transport; credentialRefs: typeof toolConnections.$inferSelect.credentialRefs; credentialSecretRefs: typeof toolConnections.$inferSelect.credentialSecretRefs; }> = []; @@ -168,6 +170,7 @@ export function accessService(db: Db) { }).from(connectionGrants).where(eq(connectionGrants.companyId, companyId)), tx.select({ id: toolConnections.id, + transport: toolConnections.transport, credentialRefs: toolConnections.credentialRefs, credentialSecretRefs: toolConnections.credentialSecretRefs, }).from(toolConnections).where(eq(toolConnections.companyId, companyId)), @@ -324,6 +327,7 @@ export function accessService(db: Db) { && grant.credentialSecretRefs.length > 0 && grant.credentialSecretRefs.every((ref) => !connectionSecretIdsToRemove.has(ref.secretId)), ); + if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); if (!hasUnaffectedActiveGrant) await notifyToolConnectionHealth(tx, companyId); await tx.update(toolConnections).set({ credentialRefs, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 1cc8a7aff6a..5059e0dddec 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1,3 +1,7 @@ +import { notifyChatCommandRegistration } from "./chat-channels/command-work-notifications.js"; +import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; +import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { notifyChatConnectionHealth } from "./chat-channels/connection-health-notifications.js"; import { createChatSdkMessageReceiver } from "./chat-channels/runtime/sdk-messages.js"; import { admitDiscordSdkMessage } from "./chat-channels/discord/messages.js"; @@ -889,7 +893,15 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { authenticateGitHubWebhook } = createGitHubWebhookAuthentication(db, { endpointRecord, runtimeContextForRecord, withCredentialMutationLease, resolveCredentials, reconcileGitHubWebhookRepository, stageGitHubWebhookIngress, processGitHubWebhookIngress, githubIngressAcceptedResponse, scheduleMessageProcessing, backgroundMessageTasks, redactError, get githubWebhookAuthenticationBarrier() { return options.githubWebhookAuthenticationBarrier; }, get deferWebhookProcessing() { return options.deferWebhookProcessing; } }); const { inspectSlackWebhookScope, recordSlackWebhookCallback } = createSlackWebhookCallbacks(db, { resolveCredentials, runtimeCallbackEndpoint, getWebhookPublicBaseUrl, scheduleMessageProcessing, slackRegistration: { notifyVerified: (endpointId) => slackRegistration.notifyVerified(endpointId) } }); - const { reconcileDiscordRuntime } = createDiscordRuntimeReconciliation(db, { reconcileDiscordCommands, endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent }); + const { reconcileDiscordRuntime } = createDiscordRuntimeReconciliation(db, { endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent }); + + function registerCommandWork(coordinator: ReturnType, canRun: () => boolean) { + return registerDiscordCommandWork(coordinator, discordCommandSchedule(db, async endpointId => { + await reconcileDiscordCommands(endpointId); + const instance = runtime.get(endpointId); + if (instance) await reconcileDiscordRuntime(endpointId, instance); + }), () => !shuttingDown && canRun()); + } async function runtimeFor( endpoint: EndpointRow, @@ -2203,6 +2215,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? "Started a conversation by mentioning the agent." : "Sent an empty message."); let comment!: Awaited>; + await notifyChatCommandRegistration(taskTx, taskEndpoint); await taskTx .update(chatEndpoints) .set({ @@ -3461,6 +3474,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor }); const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend }); return { + registerCommandWork, voice, authorizeFastResponse, slackRegistration, diff --git a/server/src/services/chat-channels/command-work-notifications.ts b/server/src/services/chat-channels/command-work-notifications.ts new file mode 100644 index 00000000000..e8f5812d4dd --- /dev/null +++ b/server/src/services/chat-channels/command-work-notifications.ts @@ -0,0 +1,18 @@ +import type { ChatProvider, ToolConnectionTransport } from "@paperclipai/shared"; +import { notifyDiscordCommandRegistration } from "./discord/command-work-notifications.js"; + +export async function notifyChatCommandRegistration(transaction: object, endpoint: { id: string; provider: ChatProvider }) { + if (endpoint.provider === "discord") { + await notifyDiscordCommandRegistration(transaction, { kind: "endpoint", id: endpoint.id }); + } +} + +export async function notifyChatConnectionCommandRegistration(transaction: object, connectionId: string, transport: ToolConnectionTransport = "chat_sdk") { + if (transport === "chat_sdk") { + await notifyDiscordCommandRegistration(transaction, { kind: "connection", id: connectionId }); + } +} + +export async function notifyChatCompanyCommandRegistration(transaction: object, companyId: string) { + await notifyDiscordCommandRegistration(transaction, { kind: "company", id: companyId }); +} diff --git a/server/src/services/chat-channels/credentials.ts b/server/src/services/chat-channels/credentials.ts index b31810ed7d9..667593066b9 100644 --- a/server/src/services/chat-channels/credentials.ts +++ b/server/src/services/chat-channels/credentials.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "./command-work-notifications.js"; import { prepareSlackInventory } from "./slack/inventory.js"; import { prepareGitHubInventory } from "./github/inventory.js"; import { prepareDiscordInventory } from "./discord/inventory.js"; @@ -315,6 +316,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { code: "chat_endpoint_credentials_changed", }); } + await notifyChatCommandRegistration(tx, endpoint); await tx .delete(companySecretBindings) .where( @@ -411,6 +413,7 @@ export function createChatCredentials(db: Db, options: ChatCredentialsOptions) { .then((rows) => rows[0] ?? null); if (!current) throw notFound("Chat connection not found"); refs = current.refs; + await notifyChatCommandRegistration(tx, endpoint); await tx .delete(companySecretBindings) .where( diff --git a/server/src/services/chat-channels/discord/command-work.ts b/server/src/services/chat-channels/discord/command-work.ts index d8140d7aeb9..dabffa73bc6 100644 --- a/server/src/services/chat-channels/discord/command-work.ts +++ b/server/src/services/chat-channels/discord/command-work.ts @@ -1,3 +1,4 @@ +import { takeCompanyWorkBatch } from "../../company-work-batch.js"; import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; import type { CommandRegistrationDeadline, discordCommandSchedule } from "./command-schedule.js"; @@ -27,6 +28,7 @@ export function registerDiscordCommandWork( const failures = new Map(); const retryMs = 30_000; let earliest = 0; + let lastCompany: string | null = null; function next() { let at: number | null = dirty.size ? earliest : null; for (const entry of deadlines.values()) { @@ -66,8 +68,10 @@ export function registerDiscordCommandWork( const due = [...deadlines.values()] .filter(entry => !unavailable.some(scope => matches(entry, scope)) && Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) - .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, 25); - for (const entry of due) { + .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); diff --git a/server/src/services/chat-channels/discord/gateway-events.ts b/server/src/services/chat-channels/discord/gateway-events.ts index f21c78cc11f..47deeff7941 100644 --- a/server/src/services/chat-channels/discord/gateway-events.ts +++ b/server/src/services/chat-channels/discord/gateway-events.ts @@ -1,3 +1,4 @@ +import { notifyDiscordCommandRegistration } from "./command-work-notifications.js"; import type { EndpointRow, DbTransaction, RuntimeContext, LifecycleRuntimeFence, InternalSetupState } from "../runtime-types.js"; import { notifyChatEndpointWork } from "../../chat-work-notifications.js"; @@ -129,6 +130,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents } const now = new Date(); if (event.type === "connecting") { + if (current.status === "attention") await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); await tx .update(chatEndpoints) .set({ @@ -157,6 +159,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents if (current.status === "attention" && !gatewayOwnedAttention) return; const status = current.setup.step === "complete" ? "active" : "verifying"; + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); if (status !== current.status) await notifyChatEndpointWork(tx, current.connectionId); await tx .update(chatEndpoints) @@ -186,6 +189,7 @@ export function createDiscordGatewayEvents(db: Db, options: DiscordGatewayEvents return; } + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: current.id }); const detail = discordGatewayFailureDetail(event); const transientGuildOutage = event.type === "guild_unavailable"; const terminal = diff --git a/server/src/services/chat-channels/discord/runtime-reconciliation.ts b/server/src/services/chat-channels/discord/runtime-reconciliation.ts index 8a59d6576b2..bd2305ae72f 100644 --- a/server/src/services/chat-channels/discord/runtime-reconciliation.ts +++ b/server/src/services/chat-channels/discord/runtime-reconciliation.ts @@ -7,11 +7,9 @@ import type { Db } from "@paperclipai/db"; import type { EndpointRow, EndpointRecord, RuntimeContext } from "../runtime-types.js"; import type { ChatRuntime } from "../../chat-runtime.js"; import type { ChatEndpointRuntime } from "../../voice/voice-runtime.js"; -import type { createDiscordCommands } from "./commands.js"; import type { createChatMessageLifecycle } from "../inbound/lifecycle.js"; import type { ToolCredentialSecretRef } from "@paperclipai/shared"; type ReconciliationOptions = { - reconcileDiscordCommands: ReturnType["reconcileDiscordCommands"]; endpointRecord: (endpointId: string) => Promise; runtimeContexts: WeakMap; invalidateRuntime: (endpointId: string) => Promise; @@ -22,9 +20,8 @@ type ReconciliationOptions = { discordGatewayRuntimeIsCurrent: (endpointId: string, context: RuntimeContext) => boolean; }; export function createDiscordRuntimeReconciliation(db: Db, options: ReconciliationOptions) { - const { reconcileDiscordCommands, endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent } = options; + const { endpointRecord, runtimeContexts, invalidateRuntime, runtimeFor, runtime, runtimeCallbackEndpoint, credentialFingerprint, discordGatewayRuntimeIsCurrent } = options; async function reconcileDiscordRuntime(endpointId: string, instance: ChatEndpointRuntime) { - await reconcileDiscordCommands(endpointId); const afterCommands = await endpointRecord(endpointId); const desiredCommands = afterCommands?.endpoint.capabilities.slashCommands && diff --git a/server/src/services/chat-channels/interactions/callbacks.ts b/server/src/services/chat-channels/interactions/callbacks.ts index 04c1f647523..7eb531ae29b 100644 --- a/server/src/services/chat-channels/interactions/callbacks.ts +++ b/server/src/services/chat-channels/interactions/callbacks.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "../command-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { createDiscordQuestionForms } from "../discord/question-forms.js"; @@ -411,6 +412,7 @@ export function createChatInteractionCallbacks(db: Db, options: InteractionCallb ["active"], ); if (currentEndpoint) { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/provider-effects.ts b/server/src/services/chat-channels/provider-effects.ts index 4b15a1b769d..1273c07c59b 100644 --- a/server/src/services/chat-channels/provider-effects.ts +++ b/server/src/services/chat-channels/provider-effects.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "./command-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { createTelegramCallbackNoticeAuthority } from "./telegram/callback-notices.js"; @@ -317,6 +318,7 @@ export function createProviderEffects(db: Db, options: ProviderEffectsOptions) { ); } if (input.disposition.kind === "endpoint_attention" && currentEndpoint) { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/publications/failures.ts b/server/src/services/chat-channels/publications/failures.ts index 84184bb8633..ae2580dfb3f 100644 --- a/server/src/services/chat-channels/publications/failures.ts +++ b/server/src/services/chat-channels/publications/failures.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "../command-work-notifications.js"; import { notifyChatConnectionHealth } from "../connection-health-notifications.js"; import { isOutboundAttachmentHydrationError, isOutboundAttachmentValidationError } from "../outbound-attachments.js"; @@ -189,6 +190,7 @@ export function createChatPublicationFailures(db: Db, options: PublicationFailur : null; if (disposition.kind === "endpoint_attention" && currentEndpoint) { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-channels/resource-reconciliation.ts b/server/src/services/chat-channels/resource-reconciliation.ts index 3658dbb3169..98bfb5822d4 100644 --- a/server/src/services/chat-channels/resource-reconciliation.ts +++ b/server/src/services/chat-channels/resource-reconciliation.ts @@ -1,3 +1,4 @@ +import { notifyChatCommandRegistration } from "./command-work-notifications.js"; import { notifyChatConnectionHealth } from "./connection-health-notifications.js"; import { imessageResourceParticipants } from "./imessage-photon/resources.js"; import { createGitHubResources, reconcileGitHubRepositoryCoordinates } from "./github/resources.js"; @@ -509,6 +510,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc : currentEndpoint.status === "attention" ? "verifying" : currentEndpoint.status; + await notifyChatCommandRegistration(tx, currentEndpoint); if (status !== currentEndpoint.status) await notifyChatEndpointWork(tx, currentEndpoint.connectionId); await tx .update(chatEndpoints) @@ -538,6 +540,7 @@ export function createChatResourceReconciliation(db: Db, options: ResourceReconc .where(eq(toolConnections.id, currentEndpoint.connectionId)); } } else { + await notifyChatCommandRegistration(tx, currentEndpoint); await tx .update(chatEndpoints) .set({ diff --git a/server/src/services/chat-discord-command-registration-store.test.ts b/server/src/services/chat-discord-command-registration-store.test.ts index db304d5ff17..8960f7a0bf1 100644 --- a/server/src/services/chat-discord-command-registration-store.test.ts +++ b/server/src/services/chat-discord-command-registration-store.test.ts @@ -805,4 +805,40 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => { expect(coordinator!.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); }); + it("retains a committed eligibility change while the provider attempt is in flight", async () => { + const f = await fixture(); await f.reconcile(); await f.due(); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + f.beforeResponse(async method => { if (method === "GET") await held; }); + const starting = scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + try { + await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST", "GET"])); + await db.transaction(async tx => { + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: f.scope.endpointId }); + await tx.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + }); + } finally { release(); } + const { errors } = await starting; + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await sqlQuiet(120_000); + expect(f.calls).toEqual(["GET", "POST", "GET"]); expect(errors).toEqual([]); + }); + + it("checks a quiet company's due command in the first batch despite another company's backlog", async () => { + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const entries = Array.from({ length: 30 }, (_, index) => ({ + id: `busy-${index.toString().padStart(2, "0")}`, companyId: "busy", connectionId: "busy-connection", dueAt: 0, + })); + entries.push({ id: "quiet", companyId: "quiet", connectionId: "quiet-connection", dueAt: 1 }); + const check = vi.fn(async (id: string) => ({ ...entries.find(entry => entry.id === id)!, dueAt: Date.now() + 300_000 })); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + await registerDiscordCommandWork(coordinator, { load: async () => entries, check }, () => true).ready; + expect(check).toHaveBeenCalledTimes(25); + expect(check.mock.calls.map(call => call[0])).toContain("quiet"); + expect(errors).toEqual([]); + }); + }); diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 1d8b1ffa518..cc4742d0b71 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionCommandRegistration } from "./chat-channels/command-work-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export async function notifyChatDeliveryWork(tx: object): Promise { @@ -37,4 +38,5 @@ export async function notifyChatEndpointWork(tx: object, connectionId: string): for (const queue of Object.values(CHAT_ACTION_QUEUES)) await notifyDeliveryWork(tx, queue); await notifyChatVerificationWork(tx); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatTelegramSubscriptions, connectionId); + await notifyChatConnectionCommandRegistration(tx, connectionId); } diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 87695be7363..0e19a37f349 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -1,3 +1,4 @@ +import { notifyChatCompanyCommandRegistration } from "./chat-channels/command-work-notifications.js"; import { notifyToolConnectionHealth } from "./tool-connection-health-notifications.js"; import { notifyCompanyRetentionWork } from "./retention-work-notifications.js"; import { notifyDecisionTargetWork } from "./decision-target-work.js"; @@ -119,6 +120,7 @@ export async function deleteCompany(db: Db, id: string) { await tx.delete(executionWorkspaces).where(eq(executionWorkspaces.companyId, id)); await tx.delete(projects).where(eq(projects.companyId, id)); await agentLifecycleCompanyDeletion.deleteCompanyData(tx, id); + await notifyChatCompanyCommandRegistration(tx, id); await notifyToolConnectionHealth(tx, id); const rows = await tx .delete(companies) diff --git a/server/src/services/tool-access/catalog/health.ts b/server/src/services/tool-access/catalog/health.ts index c387560eadf..c06ef95e324 100644 --- a/server/src/services/tool-access/catalog/health.ts +++ b/server/src/services/tool-access/catalog/health.ts @@ -1,3 +1,4 @@ +import { notifyChatConnectionCommandRegistration } from "../../chat-channels/command-work-notifications.js"; import { writeToolConnectionHealthChange } from "../../tool-connection-health-notifications.js"; import { assertSupportedConnection } from "../connections/setup-policy.js"; import { connectionSetupUrl, connectionReconnectUrl } from "../identifiers.js"; @@ -47,8 +48,9 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d message: string | null, ) { const now = new Date(); - const [updated] = await writeToolConnectionHealthChange(db, connection.companyId, tx => tx - .update(toolConnections) + const [updated] = await writeToolConnectionHealthChange(db, connection.companyId, async tx => { + await notifyChatConnectionCommandRegistration(tx, connection.id, connection.transport); + return tx.update(toolConnections) .set({ healthStatus: status, healthMessage: message, @@ -58,7 +60,8 @@ export function createConnectionHealth(db: Db, { now, audit, getConnectionRow, d updatedAt: now, }) .where(eq(toolConnections.id, connection.id)) - .returning()); + .returning(); + }); if (connection.transport === "local_stdio") { await db .update(toolRuntimeSlots) diff --git a/server/src/services/tool-access/connections/removal.ts b/server/src/services/tool-access/connections/removal.ts index 2fef00ff9ff..9be31478434 100644 --- a/server/src/services/tool-access/connections/removal.ts +++ b/server/src/services/tool-access/connections/removal.ts @@ -1,3 +1,4 @@ +import { notifyChatEndpointWork } from "../../chat-work-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { toConnection } from "./projection.js"; import { actorBinding } from "../actor.js"; @@ -537,6 +538,7 @@ async function removeConnection( .returning({ id: connectionTokenIssuances.id }); const archived = await db.transaction(async (tx) => { + if (connection.transport === "chat_sdk") await notifyChatEndpointWork(tx, connection.id); await notifyToolConnectionHealth(tx, connection.companyId); await tx.select({ id: toolApplications.id }).from(toolApplications) .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId)))