diff --git a/apps/hub/src/index.ts b/apps/hub/src/index.ts index e26efe0188..b6d8ae99c4 100644 --- a/apps/hub/src/index.ts +++ b/apps/hub/src/index.ts @@ -314,6 +314,7 @@ import { createDrizzleAccessPolicyStore, } from "@workbench/access-policy"; import { guardedHubApp, resolveCallerRoleNames } from "./tenant-create-guard"; +import { createTenantCreateObserver } from "./tenant-create-onboard"; import { createInMemoryNotifyDispatchStore, createSinkRegistry, @@ -359,10 +360,7 @@ import { drizzleAdapter } from "better-auth/adapters/drizzle"; import { type Context, Hono, type Next } from "hono"; import { upgradeWebSocket, websocket } from "hono/bun"; -import { - CORBITS_TOOLS_REGISTRY, - publishCorbitsToolsRegistry, -} from "@corbits/tool-registry-publish"; +import { CORBITS_TOOLS_REGISTRY } from "@corbits/tool-registry-publish"; import { readHubConfig, type HubConfig, @@ -3370,10 +3368,23 @@ export async function createHub(config: HubConfig) { sessionFor, log: (line) => log.info`${line}`, logError: (line) => log.error`${line}`, - publishToolRegistryFn: publishCorbitsToolsRegistry, }); benchProvisioner.start(); + // CL-7584: the tenant-create trigger. A 201 from the native + // `POST /api/tenants` route kicks a fire-and-forget desired-state + // reconcile for the new tenant under the creator's minted session — + // the revisit kick below and the drain above share this one + // reconciler. No durable row: the pending_seed row stays the only + // durable work item, and a kick lost to a restart is re-covered by + // the revisit kick on the tenant's next visit. The observer itself is + // composed just before the guard wrap, after every route mount: Hono + // copies routes at `.route()` time, so wrapping earlier would strand + // everything mounted after it. + const observerRef: { + current?: ReturnType; + } = {}; + const onboardingDeps: Parameters[0] = { hubUrl: config.baseUrl, defaultTenantSlug: config.defaultTenantSlug, @@ -3384,6 +3395,12 @@ export async function createHub(config: HubConfig) { credentialCipher, pendingSeedStore, benchProvisioner, + desiredStateKick: (args) => { + // Fire-and-forget; the route already decided pins are pending. + void observerRef.current + ?.kick({ tenantId: args.tenantId, cookies: args.cookies }) + .catch(() => undefined); + }, accessPolicy: { store: accessPolicyStore, envSignupMode: config.signupMode, @@ -3585,7 +3602,18 @@ export async function createHub(config: HubConfig) { : undefined; }, }; - const guardedApp = guardedHubApp(app, guardDeps); + const observer = createTenantCreateObserver( + { + api: selfApi, + hubUrl: config.baseUrl, + pushWorkflow: createGitWorkflowPusher(), + log: (line) => log.info`${line}`, + logError: (line) => log.error`${line}`, + }, + app, + ); + observerRef.current = observer; + const guardedApp = guardedHubApp(observer.app, guardDeps); const inFlight = createInFlightRequestTracker(); const servingApp = withInFlightRequestTracking(guardedApp, inFlight); @@ -3595,6 +3623,16 @@ export async function createHub(config: HubConfig) { db, close: async () => { sidecarAllocationReconciliationStopped = true; + // Let any in-flight tenant-create reconcile reach its next HTTP + // call before the server stops — the call then fails and the kick + // logs it, so a fire-and-forget reconcile never races the DB + // teardown. Bounded: a kick stuck on an already-dying connection + // must not stall shutdown (CL-7584). + observerRef.current?.stop(); + await Promise.race([ + observerRef.current?.whenIdle(), + new Promise((resolve) => setTimeout(resolve, 250)), + ]); if (sidecarAllocationReconciliationTimer !== undefined) { clearTimeout(sidecarAllocationReconciliationTimer); } @@ -3625,7 +3663,14 @@ export async function createHub(config: HubConfig) { await benchSettings.close(); await evalRuns.close(); await closeMailbox(); - await close(); + // The pool end waits on in-flight queries; a query whose socket + // died with the process must never stall shutdown, so bound it. + // (CL-7584: a fire-and-forget reconcile's request can be cut + // mid-query by this very teardown.) + await Promise.race([ + close(), + new Promise((resolve) => setTimeout(resolve, 5_000)), + ]); }, }; } diff --git a/apps/hub/src/tenant-create-onboard.ts b/apps/hub/src/tenant-create-onboard.ts new file mode 100644 index 0000000000..bbd2655761 --- /dev/null +++ b/apps/hub/src/tenant-create-onboard.ts @@ -0,0 +1,212 @@ +// CL-7584: the tenant-create trigger. Wrapped beside the access-policy +// guard (./tenant-create-guard.ts), this outer layer watches the native +// `POST /api/tenants` route: a 201 means a real tenant now exists that +// should converge onto the tenant desired-state document. The reconcile +// runs fire-and-forget under the creator's own session — the cookies +// that made the create are replayed, so no extra session is minted and +// the kick never touches the DB directly — so a fresh tenant gets Myra +// and the core pins without hub boot seeding anything. +// +// There is no durable work item here on purpose: the pending_seed row +// stays the only durable queue (a connect's credential), and a +// tenant-create kick that dies with the process is re-covered by the +// revisit kick (`POST /api/onboarding/provision`) on the next visit. +// In-process dedupe by tenantId is the same class of optimization the +// provisioner's in-flight map is: never a fact the system needs correct. +import { Hono } from "hono"; +import type { AppEnv } from "@intx/hub-api"; +import { cookiesFromHeader, type ApiCall } from "@corbits/hub-api-client"; +import { + reconcileTenantDesiredState, + resolveTenantModelSource, + resolveTenantDeployer, + TENANT_DESIRED_STATE, + type ReconcileReport, +} from "@workbench/onboarding/desired-state"; +import type { WorkflowPusher } from "@corbits/seeding"; + +export type TenantCreateOnboardDeps = { + api: ApiCall; + hubUrl: string; + pushWorkflow: WorkflowPusher; + log: (line: string) => void; + logError?: (line: string) => void; + /** + * The convergence step. Production resolves the tenant's deploy model + * from its resolved catalog plus the creator's deploy identity (owner + * principal + domain, via `resolveTenantDeployer` — a fresh native + * tenant has no pending-seed row to read those from) and delegates to + * `reconcileTenantDesiredState`; with no offerings it reports the + * workflow pins blocked (logged, never thrown). Tests replace the + * whole thing. + */ + reconcileFn?: (args: { + tenantId: string; + cookies: string[]; + }) => Promise; + /** + * Test seam for the reconcile step itself. The production default + * resolves the tenant's deploy model plus the creator's deploy identity + * and delegates to `reconcileTenantDesiredState`; tests inject a spy + * here to prove `principalId`/`tenantDomain` reach the seam without + * standing up the reconcile's full status-read surface. The blocked + * branches (no catalog offerings, no deployable principal) return or + * throw before this is ever entered. + */ + reconcileStateFn?: typeof reconcileTenantDesiredState; +}; + +export type TenantCreateObserver = { + /** The composed app: observes `POST /api/tenants` 201s, then falls + * through to the wrapped app. */ + app: Hono; + /** Kick a reconcile for one tenant directly (the revisit-kick wiring + * shares this with the observer). Deduped per tenant in-process. */ + kick(args: { tenantId: string; cookies: string[] }): Promise; + /** Stops accepting new kicks. In-flight ones are not interrupted — + * they run to their next HTTP call, which fails once the server has + * stopped, and the failure is caught and logged. Hub shutdown calls + * this before closing the DB and bounds the wait with `whenIdle`. */ + stop(): void; + /** Resolves when every in-flight kick has finished or bailed. */ + whenIdle(): Promise; +}; + +export function createTenantCreateObserver( + deps: TenantCreateOnboardDeps, + wrapped: Hono, +): TenantCreateObserver { + const logError = deps.logError ?? deps.log; + // In-process tenantId dedupe, same pattern as the provisioner's + // in-flight map: an optimization against double kicks, never a fact. + const kicked = new Set(); + const inFlight = new Set>(); + let stopped = false; + + async function runReconcile(args: { + tenantId: string; + cookies: string[]; + }): Promise { + if (stopped) return; + const cookies = args.cookies; + const reconcile = + deps.reconcileFn ?? + (async (reconcileArgs: { tenantId: string; cookies: string[] }) => { + const model = await resolveTenantModelSource( + deps.api, + reconcileArgs.cookies, + reconcileArgs.tenantId, + ); + if (model === undefined) { + // No catalog offerings yet — nothing is launchable. Report the + // workflow pins blocked; the next trigger (a connect's drain + // pass, a revisit probe) sees the pins still pending and + // re-kicks. + deps.log( + `tenant-create onboarding for ${reconcileArgs.tenantId} is blocked: no catalog offerings to deploy against yet`, + ); + return { + tenantId: reconcileArgs.tenantId, + ready: false, + pins: TENANT_DESIRED_STATE.workflows.map((pin) => ({ + name: pin.assetName, + kind: "workflow" as const, + status: "blocked" as const, + })), + } satisfies ReconcileReport; + } + // A fresh native tenant has no pending-seed row to read deploy + // identity from — resolve the creator's owner principal (minted + // in the create transaction) plus the tenant domain before + // reconciling, so the deploy lands under a real identity. + const deployer = await resolveTenantDeployer( + deps.api, + reconcileArgs.cookies, + reconcileArgs.tenantId, + ); + if (deployer === undefined) { + throw new Error( + `tenant-create onboarding for ${reconcileArgs.tenantId} has no deployable principal: the session holds no active user principal on the new tenant`, + ); + } + const reconcileState = + deps.reconcileStateFn ?? reconcileTenantDesiredState; + return reconcileState({ + api: deps.api, + cookies: reconcileArgs.cookies, + hubUrl: deps.hubUrl, + tenant: { + tenantId: reconcileArgs.tenantId, + principalId: deployer.principalId, + domain: deployer.tenantDomain, + }, + model, + pushWorkflow: deps.pushWorkflow, + log: deps.log, + }); + }); + const report = await reconcile({ tenantId: args.tenantId, cookies }); + deps.log( + `tenant-create onboarding for ${args.tenantId}: ${report.pins.length} pins, ready=${report.ready}`, + ); + } + + function kick(args: { tenantId: string; cookies: string[] }): Promise { + if (stopped || kicked.has(args.tenantId)) return Promise.resolve(); + kicked.add(args.tenantId); + const operation = runReconcile(args) + .catch((cause: unknown) => { + logError( + `tenant-create onboarding for ${args.tenantId} failed (the revisit kick or drain will cover it): ${cause instanceof Error ? cause.message : String(cause)}`, + ); + }) + .finally(() => { + kicked.delete(args.tenantId); + inFlight.delete(operation); + }); + inFlight.add(operation); + return operation; + } + + function stop(): void { + stopped = true; + } + + function whenIdle(): Promise { + return Promise.allSettled([...inFlight]).then(() => undefined); + } + + const app = new Hono(); + app.use("*", async (c, next) => { + await next(); + if ( + c.req.method !== "POST" || + c.req.path !== "/api/tenants" || + c.res.status !== 201 + ) { + return; + } + // The creator's own session cookies are replayed for the kick, so + // it acts under the same session that made the create without + // minting (or ever touching) anything of its own. + const cookies = cookiesFromHeader(c.req.header("cookie")); + if (cookies.length === 0) return; + const body = (await c.res + .clone() + .json() + .catch(() => undefined)) as + { id?: unknown; tenantId?: unknown } | undefined; + const tenantId = + typeof body?.id === "string" + ? body.id + : typeof body?.tenantId === "string" + ? body.tenantId + : undefined; + if (tenantId === undefined) return; + // Fire-and-forget: a 201 must answer immediately. + void kick({ tenantId, cookies }); + }); + app.route("/", wrapped); + + return { app, kick, stop, whenIdle }; +} diff --git a/apps/hub/test/artifact-doc-persistence.test.ts b/apps/hub/test/artifact-doc-persistence.test.ts index 7d5e030b07..94bc53851f 100644 --- a/apps/hub/test/artifact-doc-persistence.test.ts +++ b/apps/hub/test/artifact-doc-persistence.test.ts @@ -130,7 +130,20 @@ describeIfDb( text.insert(0, "edited by a co-editor"); registry.applyDocUpdate(key, Y.encodeStateAsUpdate(doc), principalId); - await new Promise((resolve) => setTimeout(resolve, 200)); + // The snapshot path is a 10ms-debounce timer followed by a real + // Postgres write — a fixed sleep flakes under load, so poll to a + // deadline for the version row instead of assuming any fixed delay. + const deadline = Date.now() + 10_000; + for (;;) { + const row = await getArtifact(artifactDb, created.id); + if (row?.version === 2) break; + if (Date.now() > deadline) { + throw new Error( + `the debounced snapshot never landed a version-2 row (last version: ${row?.version})`, + ); + } + await new Promise((resolve) => setTimeout(resolve, 25)); + } expect(errors).toEqual([]); const updated = await getArtifact(artifactDb, created.id); diff --git a/apps/hub/test/signup-genesis.test.ts b/apps/hub/test/signup-genesis.test.ts index cf465c3b8e..c202084c50 100644 --- a/apps/hub/test/signup-genesis.test.ts +++ b/apps/hub/test/signup-genesis.test.ts @@ -33,7 +33,11 @@ const closers: (() => Promise)[] = []; afterAll(async () => { let closer: (() => Promise) | undefined; while ((closer = closers.pop()) !== undefined) await closer(); -}); + // 60s, not bun's 5s default hook timeout: each closer shuts a whole + // booted hub (server close + DB pool drain) sequentially, and under CI + // load that chain exceeds 5s — the timeout is the teardown budget, not + // a symptom being hidden. +}, 60_000); function scratchUrlFor(label: string): string { const url = new URL(databaseUrl ?? "postgres://localhost:5432/unused"); diff --git a/apps/hub/test/tenant-create-onboard.test.ts b/apps/hub/test/tenant-create-onboard.test.ts new file mode 100644 index 0000000000..d83434401f --- /dev/null +++ b/apps/hub/test/tenant-create-onboard.test.ts @@ -0,0 +1,367 @@ +// CL-7584: the tenant-create trigger. A 201 from `POST /api/tenants` +// fires exactly one fire-and-forget desired-state reconcile for the new +// tenant under the creator's minted session; a 403 fires none; +// concurrent kicks for the same tenant dedupe in-process; and a tenant +// with no catalog offerings logs blocked instead of throwing. +import { describe, expect, test } from "bun:test"; +import { Hono } from "hono"; +import type { AppEnv } from "@intx/hub-api"; +import type { ApiCall } from "@corbits/hub-api-client"; +import type { WorkflowPusher } from "@corbits/seeding"; +import type { + ReconcileReport, + reconcileTenantDesiredState, +} from "@workbench/onboarding/desired-state"; +import { + createTenantCreateObserver, + type TenantCreateOnboardDeps, +} from "../src/tenant-create-onboard"; + +function harness( + overrides: Omit, "reconcileFn"> & { + reconcileFn?: TenantCreateOnboardDeps["reconcileFn"] | undefined; + reconcileStateFn?: typeof reconcileTenantDesiredState | undefined; + nativeApp?: Hono; + } = {}, +) { + const logged: string[] = []; + const reconciled: string[] = []; + let cookiesSeenValue = ""; + const native = + overrides.nativeApp ?? + new Hono().post("/api/tenants", (c) => + c.json({ id: "ten_new", name: "New" }, 201), + ); + + const { nativeApp, ...depOverrides } = overrides; + const deps = { + api: (async () => { + throw new Error("no hub calls expected with a reconcileFn stub"); + }) as unknown as ApiCall, + hubUrl: "https://hub.example.com", + pushWorkflow: (async () => ({ + outcome: "pushed", + commitSha: "a".repeat(40), + })) as unknown as WorkflowPusher, + log: (line) => logged.push(line), + reconcileFn: async (args) => { + reconciled.push(args.tenantId); + cookiesSeenValue = args.cookies.join(";"); + return { + tenantId: args.tenantId, + ready: true, + pins: [{ name: "assistant", kind: "workflow", status: "installed" }], + } satisfies ReconcileReport; + }, + ...depOverrides, + } as TenantCreateOnboardDeps; + + const cookiesSeen = () => cookiesSeenValue; + const { app, kick } = createTenantCreateObserver(deps, native); + return { app, kick, logged, reconciled, cookiesSeen }; +} + +describe("createTenantCreateObserver", () => { + test("a 201 create fires one reconcile for the new tenant", async () => { + const { app, reconciled, cookiesSeen } = harness(); + + const response = await app.request("/api/tenants", { + method: "POST", + headers: { + "content-type": "application/json", + cookie: "better-auth.session_token=creator", + }, + body: JSON.stringify({ name: "New" }), + }); + expect(response.status).toBe(201); + // The kick is fire-and-forget but its collector is synchronous up + // to the first await in the observer's own async work; yield once. + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(reconciled).toEqual(["ten_new"]); + expect(cookiesSeen()).toContain("better-auth.session_token=creator"); + }); + + test("a 403 create fires nothing", async () => { + const denied = new Hono().post("/api/tenants", (c) => + c.json({ error: { code: "signup_not_allowed" } }, 403), + ); + const { app, reconciled } = harness({ nativeApp: denied }); + + const response = await app.request("/api/tenants", { method: "POST" }); + expect(response.status).toBe(403); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(reconciled).toEqual([]); + }); + + test("concurrent creates of the same tenant dedupe to one reconcile", async () => { + let releaseFirst: (() => void) | undefined; + const firstKick = new Promise((resolve) => { + releaseFirst = resolve; + }); + let calls = 0; + const { app } = harness({ + reconcileFn: async (args) => { + calls += 1; + if (calls === 1) await firstKick; + return { + tenantId: args.tenantId, + ready: true, + pins: [], + }; + }, + nativeApp: new Hono().post("/api/tenants", (c) => + c.json({ id: "ten_same" }, 201), + ), + }); + + const request = () => + app.request("/api/tenants", { + method: "POST", + headers: { cookie: "better-auth.session_token=creator" }, + }); + const first = request(); + const second = request(); + await Promise.all([first, second]); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(calls).toBe(1); + releaseFirst?.(); + await new Promise((resolve) => setTimeout(resolve, 0)); + }); + + test("a tenant with no catalog offerings logs blocked instead of throwing", async () => { + const api = (async (method: string, path: string) => { + if (method === "GET" && path === "/api/tenants/ten_new/models") { + return { status: 200, data: [], cookies: [] }; + } + throw new Error(`stub api: unhandled ${method} ${path}`); + }) as unknown as ApiCall; + const { app, logged } = harness({ + api, + reconcileFn: + undefined as unknown as TenantCreateOnboardDeps["reconcileFn"], + }); + + await app.request("/api/tenants", { + method: "POST", + headers: { + "content-type": "application/json", + cookie: "better-auth.session_token=creator", + }, + body: JSON.stringify({ name: "New" }), + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect( + logged.some( + (line) => line.includes("ten_new") && line.includes("blocked"), + ), + ).toBe(true); + }); +}); + +// CL-7584: the production deployer path, DB-free. Past the +// model===undefined gate the observer resolves the creator's deploy +// identity (`resolveTenantDeployer`: GET /api/me/principals paginated + +// GET /api/tenants/:id) and hands principalId/tenantDomain to the +// reconcile seam — never a void-identity deploy when the session holds +// no deployable principal there. +describe("createTenantCreateObserver production deployer path", () => { + const DEPLOYER_PRINCIPAL_ID = "prn_creator"; + const TENANT_DOMAIN = "new.bench.local"; + + function principalsPayload(entries: { + status: string; + principalId?: string; + }) { + return { + data: [ + { + principalId: entries.principalId ?? DEPLOYER_PRINCIPAL_ID, + tenantId: "ten_new", + tenantName: "New", + tenantSlug: "new", + kind: "user", + status: entries.status, + roles: [], + }, + ], + nextCursor: null, + }; + } + + function productionApi(principals: { data: unknown[]; nextCursor: null }) { + const api = (async (method: string, path: string) => { + if (method === "GET" && path === "/api/tenants/ten_new/models") { + return { + status: 200, + data: [ + { + id: "mdl_0", + canonicalName: "llama3.2", + offerings: [ + { + offeringId: "off_0", + providerId: "cpv_1", + providerName: "ollama", + plugin: "openai-compatible", + priority: 0, + deploymentTags: [], + capabilities: [], + pricing: [], + }, + ], + }, + ], + cookies: [], + }; + } + if (method === "GET" && path === "/api/me/principals") { + return { status: 200, data: principals, cookies: [] }; + } + if (method === "GET" && path === "/api/tenants/ten_new") { + return { + status: 200, + data: { + id: "ten_new", + name: "New", + slug: "new", + domain: TENANT_DOMAIN, + parentId: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }, + cookies: [], + }; + } + throw new Error(`stub api: unhandled ${method} ${path}`); + }) as unknown as ApiCall; + return api; + } + + function postCreate(app: Hono) { + return app.request("/api/tenants", { + method: "POST", + headers: { + "content-type": "application/json", + cookie: "better-auth.session_token=creator", + }, + body: JSON.stringify({ name: "New" }), + }); + } + + // The kick is fire-and-forget behind the 201: the singleton lookup plus + // paginated-principal fetch settles after the response is serialized. + // Poll — never a fixed sleep — so the suite stays green when the whole + // monorepo test leg runs packages in parallel. + async function awaitSettled(done: () => boolean) { + const deadline = Date.now() + 2000; + while (!done()) { + if (Date.now() > deadline) break; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + } + + test("the creator's principalId and tenant domain reach the reconcile seam", async () => { + const seen: { + tenantId: string; + principalId: string; + domain: string; + modelDefined: boolean; + }[] = []; + const reconcileStateFn: typeof reconcileTenantDesiredState = (async ( + args, + ) => { + seen.push({ + tenantId: args.tenant.tenantId, + principalId: args.tenant.principalId, + domain: args.tenant.domain, + modelDefined: args.model !== undefined, + }); + return { + tenantId: args.tenant.tenantId, + ready: true, + pins: [], + }; + }) as typeof reconcileTenantDesiredState; + const { app, logged } = harness({ + api: productionApi(principalsPayload({ status: "active" })), + reconcileFn: + undefined as unknown as TenantCreateOnboardDeps["reconcileFn"], + reconcileStateFn, + }); + + const response = await postCreate(app); + expect(response.status).toBe(201); + await awaitSettled(() => seen.length === 1); + expect(seen).toEqual([ + { + tenantId: "ten_new", + principalId: DEPLOYER_PRINCIPAL_ID, + domain: TENANT_DOMAIN, + modelDefined: true, + }, + ]); + expect( + logged.some((line) => line.includes("no deployable principal")), + ).toBe(false); + }); + + test("a suspended principal never reaches the seam — blocked, never void-identity", async () => { + let calls = 0; + const reconcileStateFn: typeof reconcileTenantDesiredState = (async ( + args, + ) => { + calls += 1; + return { + tenantId: args.tenant.tenantId, + ready: true, + pins: [], + }; + }) as typeof reconcileTenantDesiredState; + const { app, logged } = harness({ + api: productionApi(principalsPayload({ status: "suspended" })), + reconcileFn: + undefined as unknown as TenantCreateOnboardDeps["reconcileFn"], + reconcileStateFn, + }); + + const response = await postCreate(app); + expect(response.status).toBe(201); + await awaitSettled(() => + logged.some((line) => line.includes("no deployable principal")), + ); + expect(calls).toBe(0); + expect( + logged.some((line) => line.includes("no deployable principal")), + ).toBe(true); + }); + + test("no principal on the tenant never reaches the seam — blocked, never void-identity", async () => { + let calls = 0; + const reconcileStateFn: typeof reconcileTenantDesiredState = (async ( + args, + ) => { + calls += 1; + return { + tenantId: args.tenant.tenantId, + ready: true, + pins: [], + }; + }) as typeof reconcileTenantDesiredState; + const { app, logged } = harness({ + api: productionApi({ data: [], nextCursor: null }), + reconcileFn: + undefined as unknown as TenantCreateOnboardDeps["reconcileFn"], + reconcileStateFn, + }); + + const response = await postCreate(app); + expect(response.status).toBe(201); + await awaitSettled(() => + logged.some((line) => line.includes("no deployable principal")), + ); + expect(calls).toBe(0); + expect( + logged.some((line) => line.includes("no deployable principal")), + ).toBe(true); + }); +}); diff --git a/apps/hub/test/tenant-create-shutdown.test.ts b/apps/hub/test/tenant-create-shutdown.test.ts new file mode 100644 index 0000000000..93aec0e163 --- /dev/null +++ b/apps/hub/test/tenant-create-shutdown.test.ts @@ -0,0 +1,187 @@ +// The tenant-create observer's shutdown contract: `stop()` refuses new +// kicks, `whenIdle()` waits out the in-flight ones, and the hub `close()` +// 250ms race bounds that wait so a stuck reconcile can never stall +// teardown. The observer half is DB-free (the reconcile step is injected); +// the `close()` half boots a real hub and so is DB-gated like every other +// suite that does. +import { describe, expect, test } from "bun:test"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import postgres from "postgres"; +import { Hono } from "hono"; +import type { ApiCall } from "@corbits/hub-api-client"; +import type { WorkflowPusher } from "@corbits/seeding"; +import type { HubConfig } from "../src/config.ts"; +import { createHub } from "../src/index.ts"; +import { + createTenantCreateObserver, + type TenantCreateOnboardDeps, +} from "../src/tenant-create-onboard.ts"; +import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url"; +import { setupDatabase } from "../../../scripts/db-setup"; +import { dbGate } from "../../../scripts/e2e/db-gate"; + +const unreachableApi = (async () => { + throw new Error("observer tests inject reconcileFn; the api is never called"); +}) as unknown as ApiCall; +const unreachablePusher = (async () => { + throw new Error( + "observer tests inject reconcileFn; the pusher is never called", + ); +}) as unknown as WorkflowPusher; + +function mountObserver( + reconcileFn: NonNullable, + logError: (line: string) => void = () => undefined, +) { + const wrapped = new Hono(); + wrapped.get("/ping", (c) => c.text("pong")); + return createTenantCreateObserver( + { + api: unreachableApi, + hubUrl: "http://localhost:1", + pushWorkflow: unreachablePusher, + log: () => undefined, + logError, + reconcileFn, + }, + // The wrapped app's type carries the hub's AppEnv; the observer only + // forwards through it, so a bare Hono app suffices for these tests. + wrapped as never, + ); +} + +describe("tenant-create observer shutdown", () => { + test("stop() refuses new kicks while whenIdle() waits out the in-flight one", async () => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + let runs = 0; + const observer = mountObserver(async ({ tenantId }) => { + runs += 1; + await gate; + return { tenantId, ready: true, pins: [] }; + }); + + const first = observer.kick({ tenantId: "ten_a", cookies: [] }); + // Let the kick reach the stuck reconcile before stopping. + await Bun.sleep(20); + observer.stop(); + // A kick after stop resolves without running anything. + await observer.kick({ tenantId: "ten_b", cookies: [] }); + expect(runs).toBe(1); + + // whenIdle is still waiting on the stuck kick, not resolved early. + let idle = false; + const idleWait = observer.whenIdle().then(() => { + idle = true; + }); + await Bun.sleep(20); + expect(idle).toBe(false); + + release(); + await first; + await idleWait; + expect(idle).toBe(true); + }); + + test("a failed kick settles (never rejects) so whenIdle always resolves", async () => { + const lines: string[] = []; + const observer = mountObserver( + async () => { + throw new Error("sidecar exploded mid-reconcile"); + }, + (line) => lines.push(line), + ); + await observer.kick({ tenantId: "ten_a", cookies: [] }); + await observer.whenIdle(); + expect(lines.some((line) => line.includes("ten_a"))).toBe(true); + }); +}); + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +describeIfDb("hub close() bound (CL-7584)", () => { + test("close() on an idle hub resolves well inside the teardown budget", async () => { + const scratchUrl = new URL( + databaseUrl ?? "postgres://localhost:5432/unused", + ); + const dbName = scratchUrl.pathname.replace(/^\//, ""); + scratchUrl.pathname = `/${dbName}_shutdown`; + const maintenanceUrl = new URL(scratchUrl); + maintenanceUrl.pathname = "/postgres"; + const maintenance = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + const scratchName = new URL(scratchUrl).pathname.replace(/^\//, ""); + try { + await maintenance.unsafe(`DROP DATABASE IF EXISTS "${scratchName}"`); + await maintenance.unsafe(`CREATE DATABASE "${scratchName}"`); + } finally { + await maintenance.end(); + } + await setupDatabase(scratchUrl.toString()); + + const root = mkdtempSync(path.join(tmpdir(), "hub-shutdown-")); + const staticDir = path.join(root, "static"); + mkdirSync(staticDir, { recursive: true }); + writeFileSync(path.join(staticDir, "index.html"), "shell"); + mkdirSync(path.join(root, "data"), { recursive: true }); + const server = Bun.serve({ + port: 0, + fetch: () => new Response("booting", { status: 503 }), + }); + const baseUrl = `http://localhost:${server.port}`; + const config: HubConfig = { + databaseUrl: scratchUrl.toString(), + baseUrl, + sessionSecret: "insecure-test-only-session-secret-0000", + hubDataDir: path.join(root, "data"), + hubStaticDir: staticDir, + defaultTenantSlug: "workbench", + signupRateLimit: { windowSeconds: 60, max: 5 }, + signInRateLimit: { windowSeconds: 60, max: 10 }, + socialProviders: {}, + signupMode: "open", + allowedEmailDomains: [], + allowPlaintextSecrets: true, + allowUnverifiedEmails: true, + sidecarProvisioners: [], + chatIdleReapMs: 30 * 60_000, + }; + const hub = await createHub(config); + server.reload({ fetch: hub.app.fetch }); + try { + const start = Date.now(); + await hub.close(); + const elapsed = Date.now() - start; + // Measured, not guessed: idle close took ~0.8s locally + // (2026-09-14, scratch Postgres on loopback); the assertion budget + // is 10s (generous: CI teardown of server + pool). The 250ms + // observer race inside close() bounds + // the kick wait by construction, so a stuck kick cannot stall + // teardown — but a stuck-kick-under-load timing proof is still + // open: production wiring has no injection point for a hung + // reconcile, so this covers the idle path only. + expect(elapsed).toBeLessThan(10_000); + } finally { + server.stop(true); + rmSync(root, { recursive: true, force: true }); + const cleanup = postgres(maintenanceUrl.toString(), { + max: 1, + onnotice: () => undefined, + }); + try { + await cleanup.unsafe( + `DROP DATABASE IF EXISTS "${scratchName}" WITH (FORCE)`, + ); + } finally { + await cleanup.end(); + } + } + }, 60_000); +}); diff --git a/apps/web/src/onboarding.ts b/apps/web/src/onboarding.ts index 48a8621258..5f1aa8e36b 100644 --- a/apps/web/src/onboarding.ts +++ b/apps/web/src/onboarding.ts @@ -555,12 +555,22 @@ export async function submitCredential( } } +export const OnboardingStep = type({ + name: "string", + label: "string", + status: "'present' | 'pending' | 'blocked'", +}); +export type OnboardingStep = typeof OnboardingStep.infer; + const CompleteSetupResult = type({ kind: "'ready' | 'provisioning' | 'unseeded'", "tenantId?": "string", "tenantSlug?": "string", "deployed?": "string[]", "pending?": "string[]", + // CL-7584: the doc-labeled desired-state steps the finishing-setup + // view renders while agents are still coming online. + "steps?": OnboardingStep.array(), }); export type CompleteSetupOutcome = @@ -571,6 +581,9 @@ export type CompleteSetupOutcome = readonly tenantSlug: string; /** See `CredentialOutcome.agentsPending`. */ readonly agentsPending: boolean; + /** CL-7584: the desired-state steps still pending, when the hub + * reports any; absent (collapsed) once everything is present. */ + readonly steps?: readonly OnboardingStep[]; } | { readonly kind: "unseeded" } | { @@ -610,18 +623,15 @@ export async function completeSetup(): Promise { return { kind: "error", message: FALLBACK_ERROR_MESSAGE }; } const agentsPending = parsed.pending.length > 0; + const steps = parsed.steps; + const common = { + tenantSlug: parsed.tenantSlug, + agentsPending, + ...(steps !== undefined && steps.length > 0 ? { steps } : {}), + }; return parsed.tenantId === undefined - ? { - kind: "connected", - tenantSlug: parsed.tenantSlug, - agentsPending, - } - : { - kind: "connected", - tenantId: parsed.tenantId, - tenantSlug: parsed.tenantSlug, - agentsPending, - }; + ? { kind: "connected", ...common } + : { kind: "connected", tenantId: parsed.tenantId, ...common }; } catch { return { kind: "error", message: FALLBACK_ERROR_MESSAGE }; } diff --git a/apps/web/src/pages/onboarding-page.tsx b/apps/web/src/pages/onboarding-page.tsx index 573b4433eb..5552cbe683 100644 --- a/apps/web/src/pages/onboarding-page.tsx +++ b/apps/web/src/pages/onboarding-page.tsx @@ -49,7 +49,11 @@ import { submitCredential, triggerFirstLoginProvisioning, } from "../onboarding"; -import type { CredentialProvider, CredentialProviderCard } from "../onboarding"; +import type { + CredentialProvider, + CredentialProviderCard, + OnboardingStep, +} from "../onboarding"; import { OnboardingLayout } from "../onboarding/onboarding-layout"; import type { SessionUser } from "../session"; @@ -84,7 +88,13 @@ type WizardState = readonly errorRefId?: string; } | { readonly phase: "submitting" } - | { readonly phase: "finishing-setup" }; + | { + readonly phase: "finishing-setup"; + /** CL-7584: the desired-state steps the hub reports as still + * pending or blocked, rendered under the loader until ready + * collapses them. */ + readonly steps: readonly OnboardingStep[]; + }; function ProviderCardButton({ provider, @@ -182,7 +192,8 @@ function initialWizardState(): WizardState { readOpenRouterConnectReturn(window.location.search) ?? readHuggingFaceConnectReturn(window.location.search); if (returned === null) return { phase: "provisioning" }; - if (returned.kind === "connected") return { phase: "finishing-setup" }; + if (returned.kind === "connected") + return { phase: "finishing-setup", steps: [] }; return { phase: "credential", error: returned.message }; } @@ -311,13 +322,30 @@ export function OnboardingPage({ user }: { readonly user: SessionUser }) { // creates one the first time an account has none. useEffect(() => { if (state.phase === "finishing-setup") { - void completeSetup().then((outcome) => { - if (outcome.kind === "connected") { - navigate("/"); - } else if (outcome.kind === "unseeded") { - setResumingUnseeded(true); - setState({ phase: "credential", error: null }); - } else { + let cancelled = false; + // Poll until ready: each pass renders the hub's own desired-state + // step list (CL-7584); a ready answer collapses it and hands off. + void (async () => { + for (;;) { + const outcome = await completeSetup(); + if (cancelled) return; + if (outcome.kind === "connected") { + if (!outcome.agentsPending) { + navigate("/"); + return; + } + setState({ + phase: "finishing-setup", + steps: outcome.steps ?? [], + }); + await new Promise((resolve) => setTimeout(resolve, 2000)); + continue; + } + if (outcome.kind === "unseeded") { + setResumingUnseeded(true); + setState({ phase: "credential", error: null }); + return; + } setState( outcome.refId === undefined ? { phase: "credential", error: outcome.message } @@ -327,9 +355,12 @@ export function OnboardingPage({ user }: { readonly user: SessionUser }) { errorRefId: outcome.refId, }, ); + return; } - }); - return; + })(); + return () => { + cancelled = true; + }; } runProvisioning(defaultTeamName(user)); // Mount-only: this reads `state.phase` exactly once, at the value @@ -421,6 +452,29 @@ export function OnboardingPage({ user }: { readonly user: SessionUser }) {

+ {state.steps.length > 0 && ( +
    + {state.steps.map((step) => ( +
  • + {step.label} + {/* `blocked` (sidecar unavailable, the hub keeps + retrying) reads differently from plain `pending` + (still working) so a held-up pin is visible, not an + identical spinner row. */} + {step.status === "blocked" && ( + + {" "} + — waiting, retrying… + + )} +
    • + ))} +
+ )}
diff --git a/apps/web/src/shell/workbench-list.tsx b/apps/web/src/shell/workbench-list.tsx index 3d482d8e8c..dbc3cd7269 100644 --- a/apps/web/src/shell/workbench-list.tsx +++ b/apps/web/src/shell/workbench-list.tsx @@ -331,7 +331,7 @@ function WorkbenchRow({ > {workbench.kind === "chat" ? ( -