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 9b12c208465..b8bac4cd919 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1036,3 +1036,19 @@ its own PR before removing the runtime discovery backstop. Slice 19 audits that owner in [Discord command registration](2026-10-25-discord-command-registration-sleep-audit.md). Runtime discovery, active lease renewal, voice, Teams, and every original pending inventory item remain in scope. No merges are authorized. + + +## Slice 19 foundation and health review status + +Discord registration now has a normalized saved-deadline reader, scoped store +notifications, paged eligibility discovery, and a shared-coordinator worker. +Thirty-six real-DB protocol/lifecycle cases and server typechecking pass. See +[the audit and remaining integration](2026-10-25-discord-command-registration-sleep-audit.md). +It is not wired into startup, the old runtime call is not removed, and no PR is +published for this slice yet. Complete producer wiring and consumer verification +before removing the backstop. + +Tool-health #17554 at `da4089960619d3da2112a4eed3cc7a3d97dbebf8` has passing +Actions checks and fresh Greptile 5/5 after the failed-probe warning repair. +Paperclip Review remains queued, so final acceptance is still open. Its PR body +now records 771 selected local tests and the outstanding gate. No PR was merged. 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 e021c25e01f..33e94a3d39a 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 @@ -1,8 +1,9 @@ # Discord command-registration scheduling audit -Parent: tool-health PR #17554, head `2ca12f806399cfcef1a474453301534f4b96c1ef`. -The parent has 770 selected passing local tests, passing workspace typecheck and -build, and live remote CI/review. It is not accepted or merged. This branch owns +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 one newly separated reason from the existing chat-runtime inventory. ## Evidence and boundary @@ -18,8 +19,9 @@ The command protocol and authorization already live in `chat-channels/discord/commands.ts`. Durable state is in existing `chat_actions` with kind `discord_command_registration`, owned by `chat-discord-command-registration-store.ts`. The store already has retryAt and -retryIndefinite outcomes. Registered state and an indefinite hold must not cause -periodic discovery simply because an endpoint is configured. Inspect the full +retryIndefinite outcomes. A registered receipt has an existing five-minute remote verification deadline; +retain it while eliminating intervening discovery scans. An indefinite hold +must not cause periodic discovery simply because an endpoint is configured. Inspect the full state protocol before deciding which attempted/unknown states may replay. This slice should give command registration its own committed-change/deadline @@ -42,7 +44,8 @@ or the generic scheduler. Keep direct setup/foreground calls compatible. generation/fingerprint fence and schedule the latest state after settlement. - Startup reconstruction must discover missing registration state in bounded pages, without letting early failures starve later eligible endpoints. -- Empty, inactive, registered, and indefinite-hold states must become quiet. +- Empty, inactive, and indefinite-hold states must become quiet. Registered + state must remain quiet between its existing five-minute verification deadlines. Due retries remain durable, respect Retry-After, survive restart, and retain deferred work through drain. Join admitted provider work at shutdown. - Verify normal and ambiguous provider outcomes through the existing protocol @@ -62,5 +65,66 @@ 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. -No existing backstop is removed by this audit. No implementation or passing -regression is claimed for slice 19 yet. +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. + + +## Implemented scheduling foundation, October 25 + +The store now exposes a normalized next deadline. A genuinely unowned namespace +is immediately due; valid receipts use their existing retryAt, including the +five-minute successful verification interval. Indefinite holds, malformed records, +foreign ownership and orphaned ownership tombstones remain parked. This is only +a scheduling hint: the existing protocol still rechecks current authority and +never adopts or recreates an unowned command. The shared retry parser and history +predicate keep the reader and mutation owner consistent. + +Store-owned initial claims, receipt transitions, and retry-result updates await +transaction intent before writing. They emit endpoint-scoped hints through the +existing coordinator topic. No table or provider request was added to discovery. + +`chat-channels/discord/command-schedule.ts` owns eligible Discord endpoint +selection and maps stored state to normalized deadlines. Startup/company discovery +uses UUID keyset pages of 100. Due checks reread current eligibility and saved +deadlines before calling the existing command reconciler, then reload the saved +result. Discovery checks shutdown cancellation between pages and rows. + +`command-work.ts` uses the shared coordinator. It coalesces endpoint, connection, +or company hints; retains changes arriving during a pass; keeps passive drain +checks in memory; and dispatches at most 25 due endpoints per pass. Unsettled +attempts back off from 30 seconds to five minutes without being mistaken for +successful completion. Saved future deadlines remain authoritative. A provider +attempt already admitted at shutdown is joined through its durable settlement. + +Verification: 36 real-PostgreSQL cases pass in +`chat-discord-command-registration-store.test.ts`, and server typechecking passes. +The cases include existing ownership/ambiguity/fencing checks plus saved deadlines, +indefinite holds, empty SQL quiet, quiet between verification deadlines, startup +reconstruction, disabled eligibility, nested rollback, first committed activation +during drain, no-progress backoff, and shutdown during provider I/O. + +Two fixture failures were repaired without changing production assertions. The +foreign-owner deadline case initially violated the live-bot uniqueness constraint; +it now archives the first endpoint before creating the competing endpoint. The +empty inventory test inherited uncertain transactions from earlier fault cases; +its setup now reconciles those known prior test transactions before registering +the scoped worker. That preserves the requirement that real uncertain writes +must be reconciled before an instance can become quiet. Final evidence is in +`/tmp/sleep19-cancellation-tests.log` and `/tmp/sleep19-final-types.log` on this host. + +## Required next integration + +- Wire configuration, readiness, provider failures, generic connection mutation, + connection removal, and company deletion producers to the scoped topic. Audit + all endpoint and connection writers; exempt metadata-only updates explicitly. +- Register the worker through the chat service, with existing standby, drain, + startup and shutdown admission. Preserve direct foreground setup calls. +- Remove command reconciliation from runtime reconciliation only after that + coverage exists; refresh a local runtime after successful command work through + the existing runtime owner, without reintroducing command discovery polling. +- Exercise real configuration producers and affected chat/startup consumers, then + run required workspace checks and publish the next stacked PR. Remote checks, + reviews and final reconciliation with master remain mandatory. +- Continue every remaining original inventory item and unresolved earlier PR gate. + This foundation does not finish this slice or the full goal. diff --git a/server/src/services/chat-channels/discord/command-schedule.ts b/server/src/services/chat-channels/discord/command-schedule.ts new file mode 100644 index 00000000000..2da9f367193 --- /dev/null +++ b/server/src/services/chat-channels/discord/command-schedule.ts @@ -0,0 +1,62 @@ +import { and, eq, gt, inArray, isNotNull, ne, or } from "drizzle-orm"; +import { chatEndpoints, toolConnections, type Db } from "@paperclipai/db"; +import { readDiscordCommandRegistrationDeadline } from "../../chat-discord-command-registration-store.js"; +import type { CommandRegistrationScope } from "./command-work-notifications.js"; + +export type CommandRegistrationDeadline = { + id: string; + companyId: string; + connectionId: string; + dueAt: number; +}; + +export function discordCommandSchedule(db: Db, reconcile: (endpointId: string) => Promise) { + async function load(scope?: CommandRegistrationScope, signal?: AbortSignal): Promise { + const deadlines: CommandRegistrationDeadline[] = []; + let cursor: string | undefined; + while (true) { + signal?.throwIfAborted(); + const rows = await db.select({ + id: chatEndpoints.id, + companyId: chatEndpoints.companyId, + connectionId: chatEndpoints.connectionId, + applicationId: chatEndpoints.botExternalId, + guildId: chatEndpoints.providerAccountId, + }).from(chatEndpoints).innerJoin(toolConnections, and( + eq(toolConnections.id, chatEndpoints.connectionId), + eq(toolConnections.companyId, chatEndpoints.companyId), + )).where(and( + eq(chatEndpoints.provider, "discord"), + inArray(chatEndpoints.status, ["verifying", "active", "attention"]), + eq(toolConnections.status, "active"), eq(toolConnections.enabled, true), + or(ne(chatEndpoints.status, "attention"), ne(toolConnections.healthStatus, "failed")), + isNotNull(chatEndpoints.botExternalId), isNotNull(chatEndpoints.providerAccountId), + scope?.kind === "endpoint" ? eq(chatEndpoints.id, scope.id) : undefined, + scope?.kind === "connection" ? eq(chatEndpoints.connectionId, scope.id) : undefined, + scope?.kind === "company" ? eq(chatEndpoints.companyId, scope.id) : undefined, + cursor ? gt(chatEndpoints.id, cursor) : undefined, + )).orderBy(chatEndpoints.id).limit(100); + for (const row of rows) { + signal?.throwIfAborted(); + if (!row.applicationId || !row.guildId) continue; + const dueAt = await readDiscordCommandRegistrationDeadline(db, { + companyId: row.companyId, endpointId: row.id, + applicationId: row.applicationId, guildId: row.guildId, + }); + if (dueAt !== null) deadlines.push({ id: row.id, companyId: row.companyId, connectionId: row.connectionId, dueAt }); + } + if (rows.length < 100) return deadlines; + cursor = rows[rows.length - 1].id; + } + } + async function check(id: string, signal?: AbortSignal): Promise { + const scope = { kind: "endpoint", id } as const; + const [before] = await load(scope, signal); + if (!before || before.dueAt > Date.now()) return before ?? null; + signal?.throwIfAborted(); + await reconcile(id); + const [after] = await load(scope, signal); + return after ?? null; + } + return { load, check }; +} diff --git a/server/src/services/chat-channels/discord/command-work-notifications.ts b/server/src/services/chat-channels/discord/command-work-notifications.ts new file mode 100644 index 00000000000..5c8e0c78a40 --- /dev/null +++ b/server/src/services/chat-channels/discord/command-work-notifications.ts @@ -0,0 +1,10 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../delivery-work-notifications.js"; + +export type CommandRegistrationScope = + | { kind: "endpoint"; id: string } + | { kind: "connection"; id: string } + | { kind: "company"; id: string }; + +export async function notifyDiscordCommandRegistration(transaction: object, scope: CommandRegistrationScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.chatCommandRegistration, `${scope.kind}:${scope.id}`); +} diff --git a/server/src/services/chat-channels/discord/command-work.ts b/server/src/services/chat-channels/discord/command-work.ts new file mode 100644 index 00000000000..d8140d7aeb9 --- /dev/null +++ b/server/src/services/chat-channels/discord/command-work.ts @@ -0,0 +1,88 @@ +import type { createDeliveryWorkCoordinator } from "../../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../../delivery-work-notifications.js"; +import type { CommandRegistrationDeadline, discordCommandSchedule } from "./command-schedule.js"; +import type { CommandRegistrationScope } from "./command-work-notifications.js"; + +function decodeScope(hint: string): CommandRegistrationScope | undefined { + if (hint === "*") return undefined; + const separator = hint.indexOf(":"); + const kind = hint.slice(0, separator), id = hint.slice(separator + 1); + if (kind !== "endpoint" && kind !== "connection" && kind !== "company") + throw new Error("Unknown command registration scope"); + return { kind, id }; +} + +function matches(entry: CommandRegistrationDeadline, scope?: CommandRegistrationScope) { + if (!scope) return true; + return (scope.kind === "endpoint" ? entry.id : scope.kind === "connection" ? entry.connectionId : entry.companyId) === scope.id; +} + +export function registerDiscordCommandWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + const dirty = new Set(["*"]); + const deadlines = new Map(); + const failures = new Map(); + const retryMs = 30_000; + let earliest = 0; + function next() { + let at: number | null = dirty.size ? earliest : null; + for (const entry of deadlines.values()) { + const due = Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0); + at = at === null ? due : Math.min(at, due); + } + return at; + } + return coordinator.register(DELIVERY_QUEUES.chatCommandRegistration, { + retryMs, + pausePassiveMaintenance: true, + onSettled: hints => { for (const hint of hints ?? ["*"]) dirty.add(hint); }, + run: async signal => { + if (!canRun() || Date.now() < earliest) return; + earliest = Date.now() + retryMs; + const requested = dirty.has("*") ? ["*"] : [...dirty]; + dirty.clear(); + const errors: unknown[] = []; + const unavailable: Array = []; + for (let index = 0; index < requested.length; index++) { + if (signal.aborted || !canRun()) { + for (const remaining of requested.slice(index)) dirty.add(remaining); + return; + } + const hint = requested[index]; + const scope = decodeScope(hint); + try { + const rows = await service.load(scope, signal); + for (const entry of deadlines.values()) if (matches(entry, scope)) deadlines.delete(entry.id); + for (const entry of rows) deadlines.set(entry.id, entry); + for (const id of failures.keys()) { + const entry = deadlines.get(id); + if (!entry || entry.dueAt > Date.now()) failures.delete(id); + } + } catch (error) { dirty.add(hint); unavailable.push(scope); errors.push(error); } + } + const due = [...deadlines.values()] + .filter(entry => !unavailable.some(scope => matches(entry, scope)) && + Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now()) + .sort((a, b) => a.dueAt - b.dueAt || a.id.localeCompare(b.id)).slice(0, 25); + for (const entry of due) { + if (signal.aborted || !canRun()) break; + try { + const current = await service.check(entry.id, signal); + if (current) deadlines.set(entry.id, current); else deadlines.delete(entry.id); + if (current && current.dueAt <= Date.now()) throw new Error("Discord command registration remains due"); + failures.delete(entry.id); + } catch (error) { + const count = Math.min((failures.get(entry.id)?.count ?? 0) + 1, 5); + failures.set(entry.id, { count, retryAt: Date.now() + Math.min(300_000, retryMs * 2 ** (count - 1)) }); + errors.push(error); + } + } + if (errors.length) throw new AggregateError(errors, "Discord command registration could not settle"); + }, + hasPending: async () => dirty.size > 0, + nextRunAt: next, + }); +} diff --git a/server/src/services/chat-discord-command-registration-store.test.ts b/server/src/services/chat-discord-command-registration-store.test.ts index 02f8c877475..db304d5ff17 100644 --- a/server/src/services/chat-discord-command-registration-store.test.ts +++ b/server/src/services/chat-discord-command-registration-store.test.ts @@ -1,3 +1,9 @@ +import { reconcileDatabaseWork } from "@paperclipai/db"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { discordCommandSchedule } from "./chat-channels/discord/command-schedule.js"; +import { registerDiscordCommandWork } from "./chat-channels/discord/command-work.js"; +import { notifyDiscordCommandRegistration } from "./chat-channels/discord/command-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { createHash, randomUUID } from "node:crypto"; import { afterAll, @@ -23,6 +29,7 @@ import { } from "@paperclipai/db"; import { discordPaperclipCommandDefinition } from "./chat-discord-command-registration.js"; import { + readDiscordCommandRegistrationDeadline, readRegisteredDiscordCommandRegistration, reconcileStoredDiscordCommandRegistration, type StoredDiscordCommandRegistrationOptions, @@ -55,7 +62,33 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => { await db?.$client.end(); await temporary?.cleanup(); }); - afterEach(() => vi.restoreAllMocks()); + let coordinator: ReturnType | undefined; + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + }); + async function scheduled(companyId: string, reconcile: (id: string) => Promise, canRun = () => true) { + await Promise.all(Array.from({ length: 4 }, () => db.execute(sql`select 1`))); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + await reconcileDatabaseWork(db, DELIVERY_QUEUES.chatCommandRegistration); + const service = discordCommandSchedule(db, reconcile); + const errors: unknown[] = []; + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun, canReconcile: () => true, onError: error => errors.push(error) }); + const worker = registerDiscordCommandWork(coordinator, { + ...service, + load: (scope, signal) => service.load(scope ?? { kind: "company", id: companyId }, signal), + }, canRun); + await worker.ready; + return { service, errors }; + } + async function sqlQuiet(milliseconds: number) { + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), execute = vi.spyOn(db, "execute"); + await vi.advanceTimersByTimeAsync(milliseconds); + expect(reads).not.toHaveBeenCalled(); + expect(writes).not.toHaveBeenCalled(); + expect(execute).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); execute.mockRestore(); + } async function fixture(applicationId = appId()) { const companyId = randomUUID(), @@ -628,4 +661,148 @@ suite("Discord command ownership store (real PostgreSQL, no network)", () => { expect(f.calls).toEqual(["GET", "POST"]); }, ); + it("exposes saved verification and retry deadlines without contacting the provider", async () => { + const f = await fixture(); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(0); + expect(f.calls).toEqual([]); + expect(await f.reconcile()).toMatchObject({ kind: "registered" }); + const row = await f.read(); + const due = Date.parse(String(row!.result!.retryAt)); + expect(due).toBeGreaterThan(Date.now() + 290_000); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(due); + expect(f.calls).toEqual(["GET", "POST"]); + const retriedAt = new Date(Date.now() + 601_125).toISOString(); + await db.update(chatActions).set({ result: { ...row!.result, outcome: "unavailable", retryAt: retriedAt } }) + .where(eq(chatActions.id, row!.id)); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(Date.parse(retriedAt)); + expect(f.calls).toEqual(["GET", "POST"]); + }); + it("parks indefinite and malformed retry records without inventing an earlier deadline", async () => { + const f = await fixture(); await f.reconcile(); const row = (await f.read())!; + for (const result of [ + { ...row.result, retryAt: "+275760-09-13T00:00:00.000Z", retryIndefinite: true }, + { ...row.result, retryAt: "invalid" }, + { ...row.result, schema: "unknown" }, + ]) { + await db.update(chatActions).set({ result }).where(eq(chatActions.id, row.id)); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBeNull(); + } + expect(f.calls).toEqual(["GET", "POST"]); + }); + it("does not schedule ownership tombstones or another endpoint's registration", async () => { + const f = await fixture(); await f.reconcile(); + await db.update(chatEndpoints).set({ status: "archived" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + const other = await fixture(f.scope.applicationId); + expect(await readDiscordCommandRegistrationDeadline(db, other.scope)).toBeNull(); + await db.delete(chatActions).where(eq(chatActions.id, (await f.read())!.id)); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBeNull(); + expect(await readDiscordCommandRegistrationDeadline(db, other.scope)).toBeNull(); + expect(other.calls).toEqual([]); + }); + it("retains the stored retry deadline for an uncertain provider write", async () => { + const f = await fixture(); + f.beforeResponse(async method => { if (method === "POST") throw new Error("lost response"); }); + expect(await f.reconcile()).toMatchObject({ kind: "unknown" }); + const row = (await f.read())!; + expect(row.payload.registration).toMatchObject({ phase: "attempted" }); + expect(await readDiscordCommandRegistrationDeadline(db, f.scope)).toBe(Date.parse(String(row.result!.retryAt))); + expect(f.calls).toEqual(["GET", "POST"]); + }); + + it("leaves no deadline or SQL after an empty registration inventory", async () => { + const reconcile = vi.fn(); + const { errors } = await scheduled(randomUUID(), reconcile); + expect(coordinator!.nextWakeAt()).toBeNull(); + await sqlQuiet(180_000); + expect(reconcile).not.toHaveBeenCalled(); expect(errors).toEqual([]); + }); + it("reconstructs a registered verification deadline and stays quiet until it is due", async () => { + const f = await fixture(); await f.reconcile(); + const due = Date.parse(String((await f.read())!.result!.retryAt)); + const { errors } = await scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + expect(coordinator!.nextWakeAt()).toBe(due); + await sqlQuiet(120_000); + expect(f.calls).toEqual(["GET", "POST"]); + await vi.advanceTimersByTimeAsync(due - Date.now()); + await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST", "GET"])); + await vi.waitFor(async () => expect(Date.parse(String((await f.read())!.result!.retryAt))).toBeGreaterThan(due)); + expect(errors).toEqual([]); + }); + it("keeps indefinite holds quiet and respects disabled endpoint eligibility", async () => { + const f = await fixture(); await f.reconcile(); const row = (await f.read())!; + await db.update(chatActions).set({ result: { ...row.result, retryIndefinite: true } }).where(eq(chatActions.id, row.id)); + const reconcile = vi.fn(); + const { service, errors } = await scheduled(f.scope.companyId, reconcile); + await sqlQuiet(600_000); expect(coordinator!.nextWakeAt()).toBeNull(); + await db.update(chatActions).set({ result: { ...row.result, retryAt: new Date(0).toISOString() } }).where(eq(chatActions.id, row.id)); + await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + expect(await service.check(f.scope.endpointId)).toBeNull(); + expect(reconcile).not.toHaveBeenCalled(); expect(errors).toEqual([]); + }); + it("honors outer rollback and wakes from committed endpoint changes after drain", async () => { + const f = await fixture(); + await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + let draining = false; + const { errors } = await scheduled(f.scope.companyId, async () => { await f.reconcile(); }, () => !draining); + const activate = async (tx: typeof db) => { + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: f.scope.endpointId }); + await tx.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, f.scope.endpointId)); + }; + await expect(db.transaction(async tx => { + await tx.transaction(async nested => activate(nested as unknown as typeof db)); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.calls).toEqual([]); + throw new Error("abort"); + })).rejects.toThrow("abort"); + await vi.advanceTimersByTimeAsync(60_000); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeNull()); + await sqlQuiet(60_000); + draining = true; + await db.transaction(async tx => activate(tx as unknown as typeof db)); + await sqlQuiet(90_000); expect(f.calls).toEqual([]); + draining = false; + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(f.calls).toEqual(["GET", "POST"])); + expect(errors).toEqual([]); + }); + it("rechecks persisted eligibility before a due provider call", async () => { + const f = await fixture(); await f.reconcile(); + const { service, errors } = await scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + await f.due(); + await db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, + (await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, f.scope.endpointId)))[0].connectionId)); + expect(await service.check(f.scope.endpointId)).toBeNull(); + expect(f.calls).toEqual(["GET", "POST"]); expect(errors).toEqual([]); + }); + + it("backs off an attempt that cannot advance its durable state", async () => { + const f = await fixture(); + const reconcile = vi.fn(async () => {}); + const { errors } = await scheduled(f.scope.companyId, reconcile); + expect(reconcile).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(reconcile).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(errors).toHaveLength(2)); + await sqlQuiet(30_000); + expect(reconcile).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(reconcile).toHaveBeenCalledTimes(3)); + expect(await f.read()).toBeUndefined(); + }); + it("joins an admitted provider attempt during shutdown", async () => { + const f = await fixture(); + let release!: () => void; + const held = new Promise(resolve => { release = resolve; }); + f.beforeResponse(async method => { if (method === "GET") await held; }); + const starting = scheduled(f.scope.companyId, async () => { await f.reconcile(); }); + await vi.waitFor(() => expect(f.calls).toEqual(["GET"])); + let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await Promise.resolve(); expect(stopped).toBe(false); + release(); await stopping; const { errors } = await starting; + expect(stopped).toBe(true); + expect(await f.read()).toMatchObject({ status: "processed" }); + expect(coordinator!.nextWakeAt()).toBeNull(); expect(errors).toEqual([]); + }); + }); diff --git a/server/src/services/chat-discord-command-registration-store.ts b/server/src/services/chat-discord-command-registration-store.ts index 22123da3bbe..749b2d93e5c 100644 --- a/server/src/services/chat-discord-command-registration-store.ts +++ b/server/src/services/chat-discord-command-registration-store.ts @@ -1,3 +1,4 @@ +import { notifyDiscordCommandRegistration } from "./chat-channels/discord/command-work-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { @@ -66,6 +67,19 @@ function safeResult(outcome: string, seconds?: number) { return { schema: resultSchema, outcome, ...retrySchedule(outcome, seconds) }; } +function registrationRetryAt(result: typeof chatActions.$inferSelect["result"]): number | null { + if (result?.schema !== resultSchema || typeof result.retryAt !== "string") return null; + const at = Date.parse(result.retryAt); + return Number.isFinite(at) ? at : null; +} + +async function hasRegistrationHistory(database: Database, applicationId: string): Promise { + const [history] = await database.select({ id: chatActions.id }).from(chatActions).where(and( + eq(chatActions.kind, kind), eq(chatActions.providerActionId, actionKey(applicationId)), + )).limit(1); + return history !== undefined; +} + async function stored( database: Database, scope: DiscordCommandRegistrationScope, @@ -136,6 +150,21 @@ export async function readRegisteredDiscordCommandRegistration( : null; } +export async function readDiscordCommandRegistrationDeadline( + database: Database, + scope: DiscordCommandRegistrationScope, +): Promise { + const current = await stored(database, scope, false, true); + if (current) { + if (current.row.result?.retryIndefinite === true) return null; + return registrationRetryAt(current.row.result); + } + const [owner] = await database.select({ applicationId: chatDiscordCommandOwners.applicationId }) + .from(chatDiscordCommandOwners).where(eq(chatDiscordCommandOwners.applicationId, scope.applicationId)); + if (owner || await hasRegistrationHistory(database, scope.applicationId)) return null; + return 0; +} + export async function reconcileStoredDiscordCommandRegistration( db: Db, options: StoredDiscordCommandRegistrationOptions, @@ -169,17 +198,8 @@ export async function reconcileStoredDiscordCommandRegistration( .where(eq(chatDiscordCommandOwners.applicationId, scope.applicationId)) .for("update"); if (!priorOwner) { - const [history] = await tx - .select({ id: chatActions.id }) - .from(chatActions) - .where( - and( - eq(chatActions.kind, kind), - eq(chatActions.providerActionId, actionKey(scope.applicationId)), - ), - ) - .limit(1); - if (history) deny(); // No adoption of history lacking the global claim. + if (await hasRegistrationHistory(tx, scope.applicationId)) deny(); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: scope.endpointId }); const actionId = randomUUID(); const inserted = await tx .insert(chatDiscordCommandOwners) @@ -209,11 +229,8 @@ export async function reconcileStoredDiscordCommandRegistration( const current = await stored(tx, scope, true, true); if (!current) deny(); // In particular, never recreate an orphaned action. const result = current.row.result; - const due = - result?.schema === resultSchema && typeof result.retryAt === "string" - ? Date.parse(result.retryAt) - : NaN; - if (!Number.isFinite(due)) deny(); + const due = registrationRetryAt(result); + if (due === null) deny(); return { state: current.state, deferred: @@ -253,6 +270,7 @@ export async function reconcileStoredDiscordCommandRegistration( await db.transaction(async (tx) => { const current = await check(tx); if (!same(current.state, expected)) deny(); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: scope.endpointId }); const result = safeResult( next.phase === "registered" ? "registered" : "unknown", ); @@ -272,6 +290,7 @@ export async function reconcileStoredDiscordCommandRegistration( }); await db.transaction(async (tx) => { const current = await check(tx); + await notifyDiscordCommandRegistration(tx, { kind: "endpoint", id: scope.endpointId }); await tx .update(chatActions) .set({ diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 878a0072ac0..017e83651c6 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -40,6 +40,7 @@ export const DELIVERY_QUEUES = { chatGitHubPublications: "chat-github-publications", chatFailedRunRetries: "chat-failed-run-retries", chatTelegramMaintenance: "chat-telegram-maintenance", + chatCommandRegistration: "chat-command-registration", chatTelegramSubscriptions: "chat-telegram-subscriptions", chatSessionSyncs: "chat-session-syncs", chatVerification: "chat-verification",