fix(server): wake Discord registration from committed changes and deadlines

Replace command maintenance inside runtime polling with scoped producer wakes, saved verification deadlines, and fair bounded dispatch. Preserve current credential and namespace authority, foreground setup, runtime continuity, and shutdown settlement.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-25 01:13:08 -05:00
1 parent e77979d4fc
commit 3755d7cace
21 files changed
+330 -22

No files matched your search

+100
View File
@@ -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.
@@ -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.
@@ -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.
@@ -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(
+1
View File
@@ -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();
+4
View File
@@ -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,
+15 -1
View File
@@ -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<typeof createDeliveryWorkCoordinator>, 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<ReturnType<typeof issuesSvc.addComment>>;
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,
@@ -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 });
}
@@ -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(
@@ -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<string, { count: number; retryAt: number }>();
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);
@@ -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 =
@@ -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<typeof createDiscordCommands>["reconcileDiscordCommands"];
endpointRecord: (endpointId: string) => Promise<EndpointRecord | null>;
runtimeContexts: WeakMap<object, RuntimeContext>;
invalidateRuntime: (endpointId: string) => Promise<boolean>;
@@ -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 &&
@@ -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({
@@ -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({
@@ -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({
@@ -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({
@@ -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<void>(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([]);
});
});
@@ -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<void> {
@@ -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);
}
+2
View File
@@ -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)
@@ -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)
@@ -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)))