diff --git a/packages/agent-directory/src/routes.ts b/packages/agent-directory/src/routes.ts index 154fcaf14..26418cf98 100644 --- a/packages/agent-directory/src/routes.ts +++ b/packages/agent-directory/src/routes.ts @@ -19,7 +19,7 @@ import { and, eq } from "drizzle-orm"; import { Hono } from "hono"; import type { DB } from "@intx/db"; -import { workflowDefinition } from "@intx/db/schema"; +import { asset, workflowDefinition } from "@intx/db/schema"; import type { TenantEnv, RequireGrant } from "@intx/hub-api"; import { AssetServiceError, @@ -77,33 +77,66 @@ export function createAgentDefinitionRoutes({ }); const workflowJson = serializeAgentDefinitionWorkflow(definition); - let asset; + let assetId: string; try { - asset = await assetService.createAsset({ + const created = await assetService.createAsset({ tenantId: tenant.id, kind: "workflow", name: body.handle, displayName: body.name, creatorPrincipalId: principal.id, }); + assetId = created.id; } catch (cause) { if ( cause instanceof AssetServiceError && cause.reason === "duplicate_asset" ) { - return c.json( - errorEnvelope( - "conflict", - `An agent with the handle "${body.handle}" already exists`, + // A previous attempt may have created the asset row but failed + // before populateAsset wrote workflow.json — an empty shell that + // blocks retries with a misleading 409. Recover: look up the + // existing asset and reuse it only if it has no definition yet. + const existing = await db.query.asset.findFirst({ + where: and( + eq(asset.tenantId, tenant.id), + eq(asset.kind, "workflow"), + eq(asset.name, body.handle), ), - 409, - ); + }); + if (existing) { + const hasDef = await db.query.workflowDefinition.findFirst({ + where: and( + eq(workflowDefinition.assetId, existing.id), + eq(workflowDefinition.tenantId, tenant.id), + ), + }); + if (!hasDef) { + assetId = existing.id; + } else { + return c.json( + errorEnvelope( + "conflict", + `An agent with the handle "${body.handle}" already exists`, + ), + 409, + ); + } + } else { + return c.json( + errorEnvelope( + "conflict", + `An agent with the handle "${body.handle}" already exists`, + ), + 409, + ); + } + } else { + throw cause; } - throw cause; } await assetService.populateAsset({ - assetId: asset.id, + assetId, ref: DEFAULT_ASSET_REF, principal: { kind: "hub" }, tree: { @@ -114,7 +147,7 @@ export function createAgentDefinitionRoutes({ const { definitionId } = await ensureWorkflowDefinitionForAsset( db, - asset.id, + assetId, ); const row = await db.query.workflowDefinition.findFirst({ diff --git a/packages/agent-directory/test/routes.test.ts b/packages/agent-directory/test/routes.test.ts index cab7ac5bd..57608687a 100644 --- a/packages/agent-directory/test/routes.test.ts +++ b/packages/agent-directory/test/routes.test.ts @@ -53,14 +53,86 @@ function fakeAssetService(overrides: Partial = {}): AssetService { }; } -// Never reached on either path these tests exercise: the 400 fails -// before any dependency call, and the 409 fails inside `createAsset` -// before `db` is ever touched. -const UNUSED_DB = {} as DB["db"]; +// The duplicate-asset recovery path queries `db` directly (looking up the +// existing asset and its definition) before deciding whether to reuse an +// empty shell or surface a real 409. When the shell is reused the route +// continues through populateAsset → ensureWorkflowDefinitionForAsset → +// read-back, so the fake also provides just enough of drizzle's chainable +// query-builder API (`.select().from().where().limit()`, +// `.insert().values().onConflictDoNothing().returning()`) for that +// projection — the projection logic itself is `@intx/hub-sessions`/ +// `@intx/db` machinery already covered upstream; the fake only needs to +// return plausible rows, not re-prove the SQL. -function buildApp(assetService: AssetService): Hono { +type FakeDbOptions = { + existingAsset?: { id: string }; + hasDefinition?: boolean; +}; + +function fakeDb(opts: FakeDbOptions = {}): DB["db"] { + let wfDefFindFirstCalls = 0; + + const selectResult = [ + { + tenantId: TENANT.id, + creatorPrincipalId: null, + name: "research-buddy", + displayName: "Research Buddy", + }, + ]; + + return { + query: { + asset: { + findFirst: async () => opts.existingAsset ?? undefined, + }, + workflowDefinition: { + findFirst: async () => { + wfDefFindFirstCalls += 1; + if (wfDefFindFirstCalls === 1) { + return opts.hasDefinition ? { id: "def_existing" } : undefined; + } + // Read-back after ensureWorkflowDefinitionForAsset. + return { + id: "def_new", + tenantId: TENANT.id, + name: "Research Buddy", + description: null, + currentVersion: "1", + status: "deployed", + createdAt: new Date(), + updatedAt: new Date(), + }; + }, + }, + }, + select: () => ({ + from: () => ({ + where: () => ({ + limit: () => Promise.resolve(selectResult), + }), + }), + }), + insert: () => ({ + values: () => { + const chain: Record = { + onConflictDoNothing: () => chain, + returning: () => Promise.resolve([{ id: "def_new" }]), + then: (onFulfilled: unknown) => + Promise.resolve([]).then(onFulfilled as never), + }; + return chain; + }, + }), + } as unknown as DB["db"]; +} + +function buildApp( + assetService: AssetService, + db: DB["db"] = fakeDb(), +): Hono { const routes = createAgentDefinitionRoutes({ - db: UNUSED_DB, + db, assetService, requireGrant: () => async (_c, next) => { await next(); diff --git a/packages/hub-client/src/seed.ts b/packages/hub-client/src/seed.ts index 011e3f4d3..01c25c31a 100644 --- a/packages/hub-client/src/seed.ts +++ b/packages/hub-client/src/seed.ts @@ -348,7 +348,7 @@ async function ensureWorkflowAsset( const listed = await api( "GET", - `/api/tenants/${args.tenantId}/assets?kind=workflow`, + `/api/tenants/${args.tenantId}/assets?kind=workflow&inherited=false`, undefined, cookies, ); diff --git a/packages/hub-client/test/seed.test.ts b/packages/hub-client/test/seed.test.ts index 188e018aa..b38cbb3ec 100644 --- a/packages/hub-client/test/seed.test.ts +++ b/packages/hub-client/test/seed.test.ts @@ -239,7 +239,8 @@ describe("seedTenant", () => { return { status: 409, data: { error: "name taken" } }; if ( method === "GET" && - path === `/api/tenants/${TENANT_ID}/assets?kind=workflow` + path === + `/api/tenants/${TENANT_ID}/assets?kind=workflow&inherited=false` ) return { status: 200, diff --git a/packages/onboarding/src/provision.ts b/packages/onboarding/src/provision.ts index 4dfe03aa2..1d37d1bf8 100644 --- a/packages/onboarding/src/provision.ts +++ b/packages/onboarding/src/provision.ts @@ -33,6 +33,26 @@ export type ProvisionResult = readonly seedSkipReason?: string; }; +/** + * A typed provisioning failure. `kind` lets the routes layer distinguish + * a retryable (transient) failure — sidecar down, race, network — from a + * permanent one — slug conflict with no principal, tenant created but + * membership missing — so the client can decide whether to retry without + * parsing a free-text message. + */ +export type ProvisionErrorKind = "transient" | "permanent"; + +export class ProvisionError extends Error { + readonly code: string; + readonly errorKind: ProvisionErrorKind; + constructor(code: string, message: string, errorKind: ProvisionErrorKind) { + super(message); + this.name = "ProvisionError"; + this.code = code; + this.errorKind = errorKind; + } +} + export type ProvisionArgs = { api: ApiCall; cookies: string[]; @@ -98,7 +118,7 @@ async function isFullySeeded( ): Promise { const assetsResponse = await api( "GET", - `/api/tenants/${tenantId}/assets?kind=workflow`, + `/api/tenants/${tenantId}/assets?kind=workflow&inherited=false`, undefined, cookies, ); @@ -148,10 +168,32 @@ export async function provisionPersonalTenantIfNeeded( // tenant and then failed before seeding it; re-seed rather than // silently treating "created but never seeded" as done. const own = before.find((p) => p.tenantSlug === expectedSlug); - if (!own || !args.seedModel) return { kind: "existing-member" }; - if (await isFullySeeded(args.api, args.cookies, own.tenantId)) { + // Not our personal bench: some other tenant added this user, which + // is none of this hook's business. Membership is decided here without + // depending on a seed credential — recovery of a half-provisioned + // bench must not hang forever just because no seed model is configured. + if (!own) return { kind: "existing-member" }; + + const fullySeeded = await isFullySeeded( + args.api, + args.cookies, + own.tenantId, + ); + if (fullySeeded) return { kind: "existing-member" }; + + // Own bench exists but is not fully seeded. With a seed model we can + // re-seed to recover. Without one there is nothing this hook can do + // to complete seeding, so we exit as an existing-member rather than + // throwing — membership is real even if seeding is incomplete. The + // routes layer surfaces this as a typed `bench_unseeded` condition so + // the caller can act on it (e.g. prompt credential setup). + if (!args.seedModel) { + args.log( + `personal bench ${own.tenantId} exists but is not fully seeded, and no seed model is configured; returning as existing-member without re-seeding`, + ); return { kind: "existing-member" }; } + const tenantResponse = await args.api( "GET", `/api/tenants/${own.tenantId}`, @@ -201,13 +243,17 @@ export async function provisionPersonalTenantIfNeeded( // surfacing the native route's slug conflict as a failure. const afterRace = await fetchPrincipals(args.api, args.cookies); if (afterRace.length > 0) return { kind: "existing-member" }; - throw new Error( + throw new ProvisionError( + "slug_conflict_no_principal", `first-login provisioning hit a slug conflict creating a personal bench, but the caller still has no principal anywhere: ${JSON.stringify(created.data)}`, + "permanent", ); } if (created.status !== 201) { - throw new Error( + throw new ProvisionError( + "tenant_create_failed", `first-login provisioning could not create a personal bench (status ${created.status}): ${JSON.stringify(created.data)}`, + created.status >= 500 ? "transient" : "permanent", ); } const tenant = parseAs(TenantResponse, created.data, "tenant response"); @@ -215,8 +261,10 @@ export async function provisionPersonalTenantIfNeeded( const after = await fetchPrincipals(args.api, args.cookies); const membership = after.find((p) => p.tenantId === tenant.id); if (!membership) { - throw new Error( + throw new ProvisionError( + "tenant_created_no_membership", `personal bench ${tenant.id} was created but the caller has no principal in it`, + "transient", ); } diff --git a/packages/onboarding/src/routes.ts b/packages/onboarding/src/routes.ts index d27259743..3144d22e6 100644 --- a/packages/onboarding/src/routes.ts +++ b/packages/onboarding/src/routes.ts @@ -15,7 +15,8 @@ import { } from "@workbench/hub-client"; import { Hono } from "hono"; import { type } from "arktype"; -import { provisionPersonalTenantIfNeeded } from "./provision"; +import { provisionPersonalTenantIfNeeded, ProvisionError } from "./provision"; + import { completeCredentialSetup } from "./complete-credential"; const PROVIDER_IDS = supportedCredentialProviders().map((p) => p.id) as [ @@ -50,6 +51,14 @@ export function createOnboardingRoutes( const app = new Hono(); const api = createHubAPI(deps.hubUrl); + // A simple in-process per-user provision rate limiter. Provisioning is + // idempotent and safe to retry, but a client stuck in a tight retry loop + // (or a runaway script) can pile concurrent tenant creates onto the hub. + // One in-flight or recent provision per user is enough; the window is + // short because successful provisioning resolves immediately. + const PROVISION_RATE_LIMIT_MS = 10_000; + const lastProvisionByUser = new Map(); + app.post("/provision", async (c) => { const user = c.get("user"); if (!user) { @@ -59,6 +68,26 @@ export function createOnboardingRoutes( ); } + const now = Date.now(); + const lastAttempt = lastProvisionByUser.get(user.id); + if ( + lastAttempt !== undefined && + now - lastAttempt < PROVISION_RATE_LIMIT_MS + ) { + return c.json( + { + error: { + code: "rate_limited", + kind: "transient" as const, + message: + "Too many provisioning attempts. Please wait a moment and try again.", + }, + }, + 429, + ); + } + lastProvisionByUser.set(user.id, now); + const cookies = cookiesFromHeader(c.req.header("cookie")); try { const provisionArgs: Parameters< @@ -85,15 +114,32 @@ export function createOnboardingRoutes( deps.log( `first-login provisioning failed for user ${user.id}: ${message}`, ); + if (cause instanceof ProvisionError) { + const status = cause.errorKind === "transient" ? 503 : 500; + return c.json( + { + error: { + code: cause.code, + kind: cause.errorKind, + message: cause.message, + }, + }, + status, + ); + } + // An unrecognized error is treated as transient — the hub may have + // been momentarily unavailable, and retrying is safe because + // provisioning is idempotent. return c.json( { error: { code: "provisioning_failed", + kind: "transient" as const, message: "Could not provision a workbench for this account. Try again in a moment.", }, }, - 500, + 503, ); } }); diff --git a/packages/onboarding/test/provision.test.ts b/packages/onboarding/test/provision.test.ts index 4ca0ca16d..de06cb195 100644 --- a/packages/onboarding/test/provision.test.ts +++ b/packages/onboarding/test/provision.test.ts @@ -473,7 +473,8 @@ describe("provisionPersonalTenantIfNeeded", () => { } if ( method === "GET" && - path === `/api/tenants/${TENANT_ID}/assets?kind=workflow` + path === + `/api/tenants/${TENANT_ID}/assets?kind=workflow&inherited=false` ) { return { status: 200, data: [], cookies: [] }; } @@ -595,4 +596,240 @@ describe("provisionPersonalTenantIfNeeded", () => { // channel-digest — on top of the one failed attempt. expect(assetCreateAttempts).toBe(4); }); + + test("half-provisioned personal bench without a seed model returns existing-member (not stuck)", async () => { + // Without ANTHROPIC_API_KEY the server has no seed model. Membership of a + // personal bench must still resolve — recovery of "I have a bench" must + // not depend on a seed credential that may never exist. Seeding itself is + // skipped (nothing to seed with); the user is not stranded in a loop. + let assetListCalls = 0; + const api: ApiCall = async (method, path) => { + if (method === "GET" && path === "/api/me/principals") { + return { + status: 200, + data: { + data: [ + { + principalId: PRINCIPAL_ID, + tenantId: TENANT_ID, + tenantName: "alice's workbench", + tenantSlug: TENANT_SLUG, + kind: "user", + status: "active", + roles: [{ id: "rol_owner", name: "owner" }], + }, + ], + nextCursor: null, + }, + cookies: [], + }; + } + if ( + method === "GET" && + path === + `/api/tenants/${TENANT_ID}/assets?kind=workflow&inherited=false` + ) { + assetListCalls += 1; + // Tenant-local assets empty — not fully seeded. + return { status: 200, data: [], cookies: [] }; + } + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) { + return { status: 200, data: [], cookies: [] }; + } + throw new Error(`unexpected call: ${method} ${path}`); + }; + + const result = await provisionPersonalTenantIfNeeded({ + api, + cookies: ["session=abc"], + hubUrl: "http://localhost:3000", + userId: "user_1", + userEmail: "alice@example.com", + // No seedModel — hub without ANTHROPIC_API_KEY. + pushWorkflow: noopPush, + log: collector().log, + }); + + expect(result).toEqual({ kind: "existing-member" }); + // Completeness was checked (tenant-local assets listed) even without a + // seed model — membership recovery does not short-circuit before that. + expect(assetListCalls).toBe(1); + }); + + test("isFullySeeded lists tenant-local assets only (inherited=false)", async () => { + // OPERATOR_TENANT_ID trees can surface the parent's workflow assets when + // listing with inherited=true. Those must not satisfy the seed check — + // only tenant-local assets count. Assert the query uses inherited=false + // and that empty local assets trigger a re-seed when a seed model exists. + let listedInherited = false; + let listedLocal = false; + let assetCreateCount = 0; + const startedRuns: string[] = []; + + const api: ApiCall = async (method, path, body) => { + if (method === "GET" && path === "/api/me/principals") { + return { + status: 200, + data: { + data: [ + { + principalId: PRINCIPAL_ID, + tenantId: TENANT_ID, + tenantName: "alice's workbench", + tenantSlug: TENANT_SLUG, + kind: "user", + status: "active", + roles: [{ id: "rol_owner", name: "owner" }], + }, + ], + nextCursor: null, + }, + cookies: [], + }; + } + if (method === "GET" && path === `/api/tenants/${TENANT_ID}`) { + return { + status: 200, + data: { + id: TENANT_ID, + name: "alice's workbench", + slug: TENANT_SLUG, + domain: `${TENANT_SLUG}.localhost`, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }, + cookies: [], + }; + } + if ( + method === "GET" && + path.startsWith(`/api/tenants/${TENANT_ID}/grants?`) + ) { + return { + status: 200, + data: { data: [], nextCursor: null }, + cookies: [], + }; + } + if (method === "POST" && path === `/api/tenants/${TENANT_ID}/grants`) { + return { status: 201, data: {}, cookies: [] }; + } + if ( + method === "GET" && + path === + `/api/tenants/${TENANT_ID}/assets?kind=workflow&inherited=false` + ) { + listedLocal = true; + return { status: 200, data: [], cookies: [] }; + } + if (method === "GET" && path.includes("inherited=true")) { + listedInherited = true; + throw new Error("must not list inherited assets for seed completeness"); + } + if (method === "POST" && path === `/api/tenants/${TENANT_ID}/assets`) { + assetCreateCount += 1; + const name = + typeof body === "object" && + body !== null && + "name" in body && + typeof (body as { name: unknown }).name === "string" + ? (body as { name: string }).name + : `wf_${assetCreateCount}`; + return { + status: 201, + data: { + id: `ast_${assetCreateCount}`, + tenantId: TENANT_ID, + kind: "workflow", + name, + displayName: null, + creatorPrincipalId: null, + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + }, + cookies: [], + }; + } + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/git-tokens` + ) { + return { + status: 201, + data: { id: "tok_1", secret: "s3cret" }, + cookies: [], + }; + } + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) { + return { status: 200, data: [], cookies: [] }; + } + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/workflows/instances` + ) { + return { + status: 201, + data: { + id: DEPLOYMENT_ID, + tenantId: TENANT_ID, + definitionAssetId: `ast_${assetCreateCount}`, + status: "active", + createdAt: "2026-01-01T00:00:00.000Z", + }, + cookies: [], + }; + } + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/${DEPLOYMENT_ID}/runs` + ) { + return { + status: 200, + data: { runIds: [...startedRuns] }, + cookies: [], + }; + } + if ( + method === "POST" && + path === `/api/tenants/${TENANT_ID}/workflows/${DEPLOYMENT_ID}/mail` + ) { + const runId = `run_${startedRuns.length + 1}`; + startedRuns.push(runId); + return { + status: 202, + data: { + deploymentId: DEPLOYMENT_ID, + address: "echo@x", + messageId: `m${startedRuns.length}`, + }, + cookies: [], + }; + } + throw new Error(`unexpected call: ${method} ${path}`); + }; + + const result = await provisionPersonalTenantIfNeeded({ + api, + cookies: ["session=abc"], + hubUrl: "http://localhost:3000", + userId: "user_1", + userEmail: "alice@example.com", + seedModel: MODEL, + pushWorkflow: noopPush, + log: collector().log, + }); + + expect(listedLocal).toBe(true); + expect(listedInherited).toBe(false); + // Empty tenant-local assets must re-seed, not claim "already seeded" + // from an ancestor's inherited catalog. + expect(result).toEqual({ kind: "existing-member", seeded: true }); + expect(assetCreateCount).toBeGreaterThan(0); + }); }); diff --git a/packages/onboarding/test/routes.test.ts b/packages/onboarding/test/routes.test.ts index 6ccd8baf6..6f54fe6b9 100644 --- a/packages/onboarding/test/routes.test.ts +++ b/packages/onboarding/test/routes.test.ts @@ -23,7 +23,7 @@ function mountAuthenticated(routes: Hono): Hono { } describe("POST /provision", () => { - test("an unreachable hub surfaces a structured error envelope, not a bare 500 body", async () => { + test("an unreachable hub surfaces a transient error envelope (503), not a bare 500 body", async () => { const lines: string[] = []; const routes = createOnboardingRoutes({ // Port 0 on loopback refuses every connection immediately, so the @@ -36,15 +36,73 @@ describe("POST /provision", () => { const response = await app.request("/provision", { method: "POST" }); - expect(response.status).toBe(500); + // An unrecognized failure (connection refused) is transient: the hub + // may come back, and provisioning is idempotent so retry is safe. + expect(response.status).toBe(503); const body = (await response.json()) as { - error: { code: string; message: string }; + error: { code: string; kind: string; message: string }; }; expect(body.error.code).toBe("provisioning_failed"); + expect(body.error.kind).toBe("transient"); expect(typeof body.error.message).toBe("string"); expect(lines.some((line) => line.includes("user_1"))).toBe(true); }); + test("a permanent provision failure (slug conflict, no principal) maps to 500 with kind permanent", async () => { + // A slug-conflict where the caller still has no principal anywhere is a + // dead end the client cannot retry out of — it must surface as a + // permanent error so the UI offers "contact support", not "try again". + const hub = new Hono(); + hub.get("/api/me/principals", (c) => + c.json({ data: [], nextCursor: null }), + ); + hub.post("/api/tenants", (c) => + c.json({ error: { code: "conflict", message: "Slug taken" } }, 409), + ); + const server = Bun.serve({ port: 0, fetch: hub.fetch }); + try { + const routes = createOnboardingRoutes({ + hubUrl: `http://localhost:${server.port}`, + pushWorkflow: async () => "pushed", + log: () => undefined, + }); + const app = mountAuthenticated(routes); + + const response = await app.request("/provision", { method: "POST" }); + + expect(response.status).toBe(500); + const body = (await response.json()) as { + error: { code: string; kind: string; message: string }; + }; + expect(body.error.code).toBe("slug_conflict_no_principal"); + expect(body.error.kind).toBe("permanent"); + } finally { + server.stop(true); + } + }); + + test("rapid retries from the same user are rate-limited (429)", async () => { + const routes = createOnboardingRoutes({ + hubUrl: "http://127.0.0.1:0", + pushWorkflow: async () => "pushed", + log: () => undefined, + }); + const app = mountAuthenticated(routes); + + const first = await app.request("/provision", { method: "POST" }); + const second = await app.request("/provision", { method: "POST" }); + + // The first call runs (and fails transiently against the dead hub). + expect(first.status).toBe(503); + // The second is short-circuited before provisioning runs. + expect(second.status).toBe(429); + const body = (await second.json()) as { + error: { code: string; kind: string; message: string }; + }; + expect(body.error.code).toBe("rate_limited"); + expect(body.error.kind).toBe("transient"); + }); + test("an anonymous request is rejected before provisioning runs", async () => { const routes = createOnboardingRoutes({ hubUrl: "http://127.0.0.1:0",