From af3442e13705aa438696fdd86f530194945f3d3c Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 15 Sep 2026 17:38:03 -0700 Subject: [PATCH 1/2] feat(hub): compose notify delivery deps with mailbox read-back Related to CL-7238 --- apps/hub/src/index.ts | 52 ++++--- apps/hub/src/notify-delivery.ts | 47 ++++++ apps/hub/test/notify-delivery.test.ts | 215 ++++++++++++++++++++++++++ 3 files changed, 290 insertions(+), 24 deletions(-) create mode 100644 apps/hub/src/notify-delivery.ts create mode 100644 apps/hub/test/notify-delivery.test.ts diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 40a4408d57..7a03cbd756 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -96,7 +96,6 @@ import type { FinalizedTurnToolCall } from "@corbits/turn-artifacts"; import { decodedOrNull } from "@corbits/url-path"; import { createInboxRoutes, - createWorkbenchMailboxDelivery, WORKBENCH_MAILBOX_VOCABULARY, } from "@corbits/inbox"; import { @@ -245,9 +244,27 @@ import { createProviderHealthStore, } from "@corbits/connections/provider-health"; import { - createInMemoryNotifyDispatchStore, - createSinkRegistry, -} from "@corbits/notify"; + applyAccessPolicyMigrations, + createAccessPolicyRoutes, + createDrizzleAccessPolicyStore, +} from "@workbench/access-policy"; +import { guardedHubApp, resolveCallerRoleNames } from "./tenant-create-guard"; +import { + loadAnchorDispatch, + resolveAnchorTenantId, + withDispatchTenantGuard, + withTenantBoundAllocationService, +} from "./dispatch-tenant-guard"; +import { createTenantCreateObserver } from "./tenant-create-onboard"; +import { createHubNotifyDeliveryDeps } from "./notify-delivery"; +import { mountMemory } from "./memory-mount"; +import { mountSkills } from "./skills-mount"; +import { + createUnavailableWorkflowMemoryRoutes, + createWorkflowMemoryRoutes, + createWorkflowMemoryStore, +} from "@corbits/memory-hub"; +import { createSkillRoutes, createWorkflowSkillRoutes } from "@corbits/skills"; import { createWorkflowAuthorRegistry, createWorkflowAuthorRoutes, @@ -553,15 +570,6 @@ export async function createHub(config: HubConfig) { config.databaseUrl, ); const mailboxBus = createInMemoryMailboxEventBus(); - // Delivery adapter for `@corbits/notify` — kept at the composition root so - // routine / approval / mention writers can inject it without the hub - // re-implementing mailbox writes. The credential-expiry sweep below is - // its first live caller; approval/run-failure/mention still have no - // writer wired to this adapter. - const mailboxDelivery = createWorkbenchMailboxDelivery({ - db: mailboxDb, - bus: mailboxBus, - }); const log = getLogger(["hub", "auth"]); // Built once, tagged, and shared by every secret-at-rest seam in this // composition root — see `hubCredentialCipher`. @@ -2426,8 +2434,8 @@ export async function createHub(config: HubConfig) { // Notify-to-reconnect for an OAuth-connected credential whose token // expired (Hugging Face today — see docs/onboarding-huggingface-connect.md): // a light periodic sweep over `@corbits/notify`'s pure - // `findDueCredentialExpiries`, mailing through the same delivery - // adapter above. `createInMemoryNotifyDispatchStore`/`createSinkRegistry()` + // `findDueCredentialExpiries`, mailing through the hub's notify delivery + // deps (`createHubNotifyDeliveryDeps`). `createInMemoryNotifyDispatchStore`/`createSinkRegistry()` // mean external sink fan-out (Slack, email) is a no-op until a sink is // registered — the mailbox row itself is what a person sees in their // inbox. Requires `@corbits/mailbox`'s and `@corbits/notify`'s own @@ -2441,15 +2449,11 @@ export async function createHub(config: HubConfig) { sidecarRouter, ), hubUrl: config.baseUrl, - notify: { - mail: mailboxDelivery, - addressing: { - inbox: (recipient) => `${recipient.principalId}@inbox.${notifyHost}`, - from: (kind) => `${kind}@notify.${notifyHost}`, - }, - dispatch: createInMemoryNotifyDispatchStore(), - sinks: createSinkRegistry(), - }, + notify: createHubNotifyDeliveryDeps({ + mailboxDb, + bus: mailboxBus, + host: notifyHost, + }), }); // Reopen a snoozed inbox item once its `until` has passed (CL-7208) — a diff --git a/apps/hub/src/notify-delivery.ts b/apps/hub/src/notify-delivery.ts new file mode 100644 index 0000000000..663f13bf0a --- /dev/null +++ b/apps/hub/src/notify-delivery.ts @@ -0,0 +1,47 @@ +// The hub's `@corbits/notify` delivery deps, composed once at the root so +// every mail writer (the credential-expiry sweep today) mails through the +// same adapter the inbox read-back repairs. Kept here instead of inline in +// `index.ts` so the composed path is a single construction site the +// crash-window suite can exercise directly. +import { + createResolveExistingMailIds, + createWorkbenchMailboxDelivery, +} from "@corbits/inbox"; +import type { MailboxDb, MailboxEventBus } from "@corbits/mailbox"; +import { + createInMemoryNotifyDispatchStore, + createSinkRegistry, + type NotifyDeliveryDeps, + type NotifyDispatchStore, + type SinkRegistry, +} from "@corbits/notify"; + +export interface CreateHubNotifyDeliveryDepsOpts { + mailboxDb: MailboxDb; + bus?: MailboxEventBus; + host: string; + dispatch?: NotifyDispatchStore; + sinks?: SinkRegistry; +} + +export function createHubNotifyDeliveryDeps( + opts: CreateHubNotifyDeliveryDepsOpts, +): NotifyDeliveryDeps { + return { + mail: createWorkbenchMailboxDelivery({ + db: opts.mailboxDb, + ...(opts.bus !== undefined ? { bus: opts.bus } : {}), + }), + addressing: { + inbox: (recipient) => `${recipient.principalId}@inbox.${opts.host}`, + from: (kind) => `${kind}@notify.${opts.host}`, + }, + dispatch: opts.dispatch ?? createInMemoryNotifyDispatchStore(), + sinks: opts.sinks ?? createSinkRegistry(), + // CL-7238: without this read-back a redelivery after a crash between + // the mail write and the dispatch enqueue still loses the dispatch. + // Resolved from the mailbox unique mail key — the same key the write + // dedupes on — so the lookup can never disagree with the delivery. + resolveExistingMailIds: createResolveExistingMailIds(opts.mailboxDb), + }; +} diff --git a/apps/hub/test/notify-delivery.test.ts b/apps/hub/test/notify-delivery.test.ts new file mode 100644 index 0000000000..35bcedf09f --- /dev/null +++ b/apps/hub/test/notify-delivery.test.ts @@ -0,0 +1,215 @@ +// CL-7238 prod closure through the hub's own composed notify deps: the +// credential-expiry sweep mails through `createHubNotifyDeliveryDeps`, so a +// crash between the mail write and the dispatch enqueue must be repaired on +// redelivery — the same crash-window pattern PR #734 proves for the bare +// adapters, here through the exact factory `apps/hub/src/index.ts` composes. +// Runs against its own scratch database, never the developer's or the +// walking-skeleton suite's. +import { afterAll, beforeAll, expect, test } from "bun:test"; +import postgres from "postgres"; + +import { createDB, schema } from "@intx/db"; +import { generateId } from "@intx/hub-common"; +import { + createInMemoryMailboxEventBus, + createMailboxDb, + listUserMailbox, +} from "@corbits/mailbox"; +import { + deliverApprovalMail, + type ApprovalNotification, + type NotifyDispatchStore, + type SinkDeliveryResult, +} from "@corbits/notify"; +import { WORKBENCH_INBOX_PRIORITIES } from "@corbits/inbox"; + +import { setupDatabase } from "../../../scripts/db-setup"; +import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url"; +import { dbGate } from "../../../scripts/e2e/db-gate"; +import { createHubNotifyDeliveryDeps } from "../src/notify-delivery"; + +function scratchUrlFor(e2eUrl: string): string { + const url = new URL(e2eUrl); + const database = url.pathname.replace(/^\//, ""); + url.pathname = `/${database}_hub_notify_delivery_test`; + return url.toString(); +} + +// Parse DATABASE_URL the same way the hub does (apps/hub/src/index.ts): an +// empty user falls through to the postgres client's OS-username default. +function dbConfigFromUrl(databaseUrl: string) { + const url = new URL(databaseUrl); + return { + host: url.hostname, + port: url.port === "" ? 5432 : Number(url.port), + user: decodeURIComponent(url.username), + password: decodeURIComponent(url.password), + database: url.pathname.replace(/^\//, ""), + }; +} + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +describeIfDb( + "hub notify delivery repairs the CL-7238 crash window on redelivery", + () => { + const scratchUrl = scratchUrlFor( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + const scratchDatabase = new URL(scratchUrl).pathname.replace(/^\//, ""); + + const tenantId = generateId("tenant"); + const principalId = generateId("principal"); + + beforeAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe( + `DROP DATABASE IF EXISTS "${scratchDatabase}"`, + ); + await maintenance.unsafe(`CREATE DATABASE "${scratchDatabase}"`); + } finally { + await maintenance.end(); + } + + // The same path the hub boots with: the platform's own migrations, + // then every installed package's, including `@corbits/notify` and + // `@corbits/mailbox`. + await setupDatabase(scratchUrl); + + const { db, close } = createDB(dbConfigFromUrl(scratchUrl)); + try { + await db.insert(schema.tenant).values({ + id: tenantId, + name: "Hub Notify Test Bench", + slug: `hub-notify-${tenantId}`, + domain: `hub-notify-${tenantId}.localhost`, + }); + await db.insert(schema.principal).values({ + id: principalId, + tenantId, + kind: "agent", + refId: "not-a-real-agent-instance", + status: "active", + }); + } finally { + await close(); + } + // `setupDatabase` applies the platform's migrations plus every + // installed package's; under `bun run test`'s cross-package + // concurrency that comfortably exceeds bun:test's default 5s hook + // timeout, so it gets an explicit one here rather than a flaky suite. + }, 30000); + + afterAll(async () => { + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await maintenance.unsafe( + `DROP DATABASE IF EXISTS "${scratchDatabase}"`, + ); + } finally { + await maintenance.end(); + } + }); + + test("a crash-window redelivery through the composed hub deps queues its dispatch", async () => { + const mailboxDb = createMailboxDb(scratchUrl); + try { + // Exactly what the hub composes: the factory's own dispatch store + // and sink registry are the prod path (no sink registered means no + // fan-out, only the mailbox row the sweep's reader sees). The test + // registers one sink so the repaired dispatch has somewhere to go, + // and faults only the dispatch enqueue to open the crash window. + const deps = createHubNotifyDeliveryDeps({ + mailboxDb: mailboxDb.db, + bus: createInMemoryMailboxEventBus(), + host: "hub.test", + }); + const delivered: SinkDeliveryResult = { status: "delivered" }; + deps.sinks.register({ + name: "always", + isEnabledFor: async () => true, + deliver: async () => delivered, + }); + const dispatch = deps.dispatch; + let crashNext = true; + const crashingDispatch: NotifyDispatchStore = { + ...dispatch, + enqueue: async (inputs) => { + if (crashNext) { + crashNext = false; + throw new Error( + "boom: crash between the mail write and the dispatch enqueue", + ); + } + await dispatch.enqueue(inputs); + }, + }; + const event: ApprovalNotification = { + kind: "approval", + approvalId: generateId("approval"), + tenantId, + runId: generateId("workflowRun"), + deploymentId: generateId("workflowRun"), + toolName: "quarantine_token", + toolArguments: { repo: "corbitsdev/workbench" }, + recipients: [{ tenantId, principalId }], + createdAt: new Date().toISOString(), + }; + + // The first attempt commits its mail row through the real adapter, + // then dies before the dispatch enqueue — the crash window. + await expect( + deliverApprovalMail({ ...deps, dispatch: crashingDispatch }, event), + ).rejects.toThrow("boom"); + + // The redelivery runs through the pristine composed deps: it + // dedupes on the mail key (no new row), resolves the committed + // row's id through the hub's read-back wiring, and queues the + // dispatch the crash swallowed. Without `resolveExistingMailIds` + // in the composed deps this queues nothing. + const redelivery = await deliverApprovalMail(deps, event); + expect(redelivery.deliveredMailboxRowIds).toEqual([]); + expect(redelivery.queuedDispatchCount).toBe(1); + + const page = await listUserMailbox(mailboxDb.db, { + tenantId, + principalId, + limit: 10, + view: "all", + priorities: WORKBENCH_INBOX_PRIORITIES, + }); + // Exactly one mail row for the redelivered event — the crash and + // the redelivery deduped onto the first attempt's row. + const matching = page.items.filter((item) => + (item.subject ?? "").includes("quarantine_token"), + ); + expect(matching).toHaveLength(1); + const mailId = matching[0]?.id; + if (mailId === undefined) + throw new Error("expected the redelivered mail row to list"); + expect(await dispatch.listFor(mailId)).toHaveLength(1); + + // A further benign redelivery repairs the same row again, and the + // dispatch store's (mail row, sink) dedupe holds it to one row. + const third = await deliverApprovalMail(deps, event); + expect(third.deliveredMailboxRowIds).toEqual([]); + expect(third.queuedDispatchCount).toBe(1); + expect(await dispatch.listFor(mailId)).toHaveLength(1); + } finally { + await mailboxDb.close(); + } + }, 15000); + }, +); From 8972500d16d264bd71ea70f445694b1cb6c67673 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 09:12:13 -0700 Subject: [PATCH 2/2] fix(hub): resolve notify-delivery rebase conflict against current main Rebase onto origin/main (post tenancy deletion) mis-resolved the notify import block, keeping references to files already removed on main (tenant-create-guard, access-policy, memory-mount, skills-mount). Restore main's direct createHubNotifyDeliveryDeps composition. Claude-Session: https://claude.ai/code/session_01RweuMUskU2nvcFhdckc7EG --- apps/hub/src/index.ts | 21 --------------------- 1 file changed, 21 deletions(-) diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index 7a03cbd756..dbf94a4411 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -243,28 +243,7 @@ import { createProviderHealthPort, createProviderHealthStore, } from "@corbits/connections/provider-health"; -import { - applyAccessPolicyMigrations, - createAccessPolicyRoutes, - createDrizzleAccessPolicyStore, -} from "@workbench/access-policy"; -import { guardedHubApp, resolveCallerRoleNames } from "./tenant-create-guard"; -import { - loadAnchorDispatch, - resolveAnchorTenantId, - withDispatchTenantGuard, - withTenantBoundAllocationService, -} from "./dispatch-tenant-guard"; -import { createTenantCreateObserver } from "./tenant-create-onboard"; import { createHubNotifyDeliveryDeps } from "./notify-delivery"; -import { mountMemory } from "./memory-mount"; -import { mountSkills } from "./skills-mount"; -import { - createUnavailableWorkflowMemoryRoutes, - createWorkflowMemoryRoutes, - createWorkflowMemoryStore, -} from "@corbits/memory-hub"; -import { createSkillRoutes, createWorkflowSkillRoutes } from "@corbits/skills"; import { createWorkflowAuthorRegistry, createWorkflowAuthorRoutes,