openvibers/openvibe.ai Commits
Generate runs can pin one routing offer; GET /api/v1/routes lists offers (#184)
Changes
@@@ -108,6 +108,7 @@ idempotent replay):| `POST /api/v1/{chat,generate,summarize,classify,extract,embed,image}` (no `enrich`), `GET /runs/:id` (the app's own runs) | `ai.app.run` for a developer-app token ([Developer apps](#developer-apps)) || `GET /api/v1/runs`, `GET /runs/:id` (+ citations, request-log metadata), `GET /runs/:id/citations` | `ai.run.read` (own runs) || `GET /api/v1/workflows\|templates\|routes[/:key]` | `ai.run.create`, `ai.workflow.manage` or `ai.usage.read` |++| `GET /api/v1/routes?operation=generate` | the same read: every route's `offers` (`{provider, model}`), plus the operation's deduplicated `offers` || `POST /api/v1/{templates,workflows,routes}/:key/versions`, `…/versions/:v/status` | `ai.workflow.manage` || `POST/PATCH /api/v1/providers…`, `…/disable\|enable\|reset`, `POST/PATCH /api/v1/models…`, `POST /api/v1/quotas`, `DELETE /api/v1/cache` | `ai.provider.manage` || `GET /api/v1/status\|usage\|quotas\|requests\|audit\|cache\|providers\|models` | `ai.usage.read` |@@@ -340,7 +341,17 @@ provider that is disabled, missing credentials, circuit-open or unable to do thefailure; a fallback that answers is recorded on the run (`fallback_used`), in the request log(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded--reason). `GET /api/v1/runs/:id` adds `explain.usage_readings` when the run has a Billing reading: per++reason).++++`GET /api/v1/routes?operation=generate` lists each route's eligible offers as `{ provider, model }`.++A first-party service holding `ai.run.create` may pin one on `POST /api/v1/generate` (and on++`POST /api/v1/runs` for `ai.generate`) with `"route": { "provider": …, "model": … }`. The run tries++that offer alone, never a fallback; a provider/model the routing does not list for `generate` is++`422 input.invalid` in words, and the caller's `ns` is still checked first (a workflow outside it is++`403 capability.namespace_denied`). Developer-app tokens (`ai.app.run`) cannot pin. Quotas, tier budgets and Billing readings are exactly as for an unpinned++run, and a pinned run is never served from the shared cache.++++`GET /api/v1/runs/:id` adds `explain.usage_readings` when the run has a Billing reading: perreading its `idempotency_key`, `state` (`queued` until Billing accepts it, `sent`, or `failed` when Billingrefused it), `attempts`, `queued_at`, `sent_at`, `rejected_at`, `next_attempt_at`, `last_error` (theoutbox's last send error, also while a queued reading is retried) and `free_allowance_used` when some@@@ -79,14 +79,73 @@ function adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops = createmplates: { list: registry.listTemplates, get: registry.getTemplate, create: registry.createTemplateVersion, one: 'template' },workflows: { list: registry.listWorkflows, get: registry.getWorkflow, create: registry.createWorkflowVersion, one: 'workflow' },};++++ /**++ * The offers one route's routing lists for an operation: the eligible provider/model candidates in the++ * order placement would try them. GET /api/v1/routes publishes these (and the operation can be asked for++ * with ?operation=generate); a generate run may pin one, and server/runs.js checks it against this same list.++ */++ function operationFor(route, requested) {++ const own = route && route.capability ? route.capability : null;++ if (!requested) return own || 'chat';++ if (!own || own === requested) return requested;++ // A chat route serves generate too; every other route keeps its own operation's offers.++ return own === 'chat' && requested === 'generate' ? requested : own;++ }++ async function routeOffers(route, operation, req) {++ if (!route || route.status !== 'active') return [];++ try {++ let target = route;++ if (route.alias_of) {++ target = await registry.resolveRoute(route.key);++ if (!target || target.disabled) return [];++ }++ return await pool.offers(target, operationFor(target, operation), { requesterType: req.principal.subject.type, requesterId: req.principal.subject.id });++ } catch (err) {++ log.warn(`[ai] offers for route ${route.key}: ${err.message}`);++ return [];++ }++ }++ r.get('/api/v1/routes', defRead, wrap(async (req) => {++ const operation = req.query.operation ? String(req.query.operation) : null;++ const routes = await registry.listRoutes({ history: req.query.history === '1', namespace: req.query.namespace });++ for (const route of routes) route.offers = await routeOffers(route, operation, req);++ // With ?operation=…, also the operation's own menu. For the default operations that is the++ // default route's list (ai.generate uses default.chat), so every offer there is pinnable for++ // that operation; an operation with no default route gets the routes' union.++ if (!operation) return { routes };++ const baseKey = operation === 'chat' || operation === 'generate' ? 'default.chat'++ : operation === 'embed' ? 'default.embedding'++ : operation === 'image' ? 'default.image' : null;++ const base = baseKey ? routes.find((x) => x.key === baseKey) : null;++ const offers = [];++ const seen = new Set();++ for (const route of (base ? [base] : routes)) for (const o of route.offers) {++ const id = `${o.provider}|${o.model || ''}`;++ if (seen.has(id)) continue;++ seen.add(id);++ offers.push(o);++ }++ return { routes, offers };++ }));++ r.get('/api/v1/routes/:key', defRead, wrap(async (req) => {++ const latest = await registry.getRoute(req.params.key, req.query.version ? Number(req.query.version) : undefined);++ if (!latest) throw new AiError(404, 'ai.not_found', `no route ${req.params.key}`);++ const versions = (await registry.listRoutes({ history: true })).filter(x => x.key === req.params.key).map(x => ({ version: x.version, status: x.status, created_by: x.created_by, created_at: x.created_at }));++ const operation = req.query.operation ? String(req.query.operation) : null;++ return { route: { ...latest, offers: await routeOffers(latest, operation, req) }, versions };++ }));++for (const [plural, k] of Object.entries(kinds)) {-- r.get(`/api/v1/${plural}`, defRead, wrap(async (req) => ({ [plural]: await k.list({ history: req.query.history === '1', namespace: req.query.namespace }) })));-- r.get(`/api/v1/${plural}/:key`, defRead, wrap(async (req) => {-- const latest = await k.get(req.params.key, req.query.version ? Number(req.query.version) : undefined);-- if (!latest) throw new AiError(404, 'ai.not_found', `no ${k.one} ${req.params.key}`);-- const versions = (await k.list({ history: true })).filter(x => x.key === req.params.key).map(x => ({ version: x.version, status: x.status, created_by: x.created_by, created_at: x.created_at }));-- return { [k.one]: latest, versions };-- }));++ if (plural !== 'routes') {++ r.get(`/api/v1/${plural}`, defRead, wrap(async (req) => ({ [plural]: await k.list({ history: req.query.history === '1', namespace: req.query.namespace }) })));++ r.get(`/api/v1/${plural}/:key`, defRead, wrap(async (req) => {++ const latest = await k.get(req.params.key, req.query.version ? Number(req.query.version) : undefined);++ if (!latest) throw new AiError(404, 'ai.not_found', `no ${k.one} ${req.params.key}`);++ const versions = (await k.list({ history: true })).filter(x => x.key === req.params.key).map(x => ({ version: x.version, status: x.status, created_by: x.created_by, created_at: x.created_at }));++ return { [k.one]: latest, versions };++ }));++ }// An edit is always a new version; the previous one stays readable.r.post(`/api/v1/${plural}/:key/versions`, defWrite, wrap(async (req, res) => {try {@@@ -16,6 +16,8 @@* POST /api/v1/runs/:id/citations attach sources/citations to a finished run (ai.run.create, owner)* POST /api/v1/{chat,generate,summarize,classify,extract,enrich,embed}* direct operations: a run of workflow ai.<op>, waited for++ * generate also takes route: {provider, model}, one offer from++ * GET /api/v1/routes?operation=generate (first-party services only)** Developer apps (ADR-014, ai.app.run): POST /api/v1/{chat,generate,summarize,classify,extract,embed}* and GET /api/v1/runs/:id also take an app token; the run is the app's project's (see auth.appOrCap,@@@ -174,9 +176,9 @@ function runsRouter({ runs, registry, auth, config, log = console }) {r.post(`/api/v1/${op}`, DIRECT_OPS.includes(`ai.${op}`) ? createApp : create, async (req, res) => {try {const body = req.body || {};-- const { idempotency_key: idem, target, attribution, on_behalf_of: obo, options, ...input } = body;++ const { idempotency_key: idem, target, attribution, on_behalf_of: obo, options, route, ...input } = body;const wait = req.query.wait != null ? Math.max(0, Math.min(Number(req.query.wait) || 0, config.runs.maxWaitMs)) : config.runs.maxWaitMs;-- await createAndRespond(req, res, { workflow: `ai.${op}`, input, idempotency_key: idem, target, attribution, on_behalf_of: obo, options }, wait);++ await createAndRespond(req, res, { workflow: `ai.${op}`, input, idempotency_key: idem, target, attribution, on_behalf_of: obo, options, route }, wait);} catch (err) { sendError(res, err, req.ov, log); }});}@@@ -55,7 +55,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now// A person's own provider keys (WS-O task 2): stored by the service holding their consent, used by their runs only.const credentials = require('./credentials').createCredentials({ db, config, clock });const projectSecrets = require('./project-secrets').createProjectSecrets({ config, env, fetchImpl });-- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, tiers, projectSecrets });++ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, tiers, projectSecrets, pool });await seed({ registry, quotas, config, env, db });const interrupted = await runs.recoverInterrupted();if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);@@@ -449,6 +449,25 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nothrow new AiError(502, 'credential.provider_failed', `the person's own provider did not answer: ${String(lastErr && lastErr.message || lastErr).slice(0, 200)}`);}++ /**++ * The offers the routing lists for one operation of one route: the eligible candidates in the++ * order placement would try them, as { provider, model }. GET /api/v1/routes publishes these and++ * a pinned run is accepted only when it names one of them.++ */++ async function offers(route, operation, ctx = {}) {++ const { explain } = await candidates(route, [operation], operation, ctx);++ const out = [];++ const seen = new Set();++ for (const c of explain.candidates) {++ if (!c || !c.eligible || !c.provider) continue;++ const id = offerId(c.provider, c.model || null);++ if (seen.has(id)) continue;++ seen.add(id);++ out.push({ provider: c.provider, model: c.model || null });++ }++ return out;++ }++async function execute(route, operation, req, ctx = {}) {if (ctx.credential) return await executeWithCredential(ctx.credential, operation, req, ctx);const features = [operation];@@@ -461,7 +480,19 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no// after that is final, never retried or handed to a fallback whose words would follow the first one's.let streamed = false;const onDelta = typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;-- const { order: list, explain } = await candidates(route, features, operation, ctx);++ let { order: list, explain } = await candidates(route, features, operation, ctx);++ // A first-party service's generate run may name one offer from GET /api/v1/routes. The run then++ // tries that offer alone: no fallback to another model, so the caller gets the model it asked for++ // or an explicit failure. server/runs.js validates the pin against this same offer list first.++ if (ctx.routePin && operation === 'generate') {++ const pin = ctx.routePin;++ const chosen = list.find((c) => c.provider === pin.provider && (pin.model == null || c.model === pin.model));++ if (!chosen) {++ throw new AiError(422, 'route.not_offered', `route ${pin.model ? `${pin.provider}/${pin.model}` : pin.provider} is not offered for generate; GET /api/v1/routes?operation=generate lists the offers`);++ }++ list = [chosen];++ explain.reasons = [`pinned to ${chosen.model ? `${chosen.provider}/${chosen.model}` : chosen.provider} for this run`, ...(explain.reasons || [])];++ }for (let i = 0; i < list.length; i++) {const c = list[i];const entry = await adapter(c.provider);@@@ -548,7 +579,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nothrow new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried, explain });}-- return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, rateCardsFor, providerStates, stats, statFor, candidates };++ return { adapter, health, resetHealth, execute, executeWithCredential, offers, priceFor, costOf, rateCardsFor, providerStates, stats, statFor, candidates };}module.exports = { createProviderPool };@@@ -73,7 +73,33 @@ function projectProviderOf(spec) {return { base_url: `${u.origin}${u.pathname.replace(/\/+$/, '')}`, model: String(spec.model), secret: String(spec.secret) };}--function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null, tiers = null, projectSecrets = null }) {++/**++ * A generate run's `route: { provider, model }`: one offer from GET /api/v1/routes. Only a first-party++ * service may ask for one, and only on the generate operation; the offer itself is checked against the++ * routing's own list later, once the run's route is resolved.++ */++const ROUTE_PIN_KEYS = new Set(['provider', 'model']);++function routePinRequest(value, principal, wf) {++ if (value == null) return null;++ if (principal.kind === 'app') throw new AiError(403, 'capability.denied', 'a developer app may not pin a route; use the routing pool');++ if (!principal.subject || principal.subject.type !== 'service') throw new AiError(403, 'capability.denied', 'only a first-party service may pin a route');++ if (!wf || wf.key !== 'ai.generate') throw new AiError(422, 'input.invalid', 'route pinning is available on the generate operation (ai.generate) only');++ if (typeof value !== 'object' || Array.isArray(value)) throw new AiError(422, 'input.invalid', 'route must be { provider, model }');++ for (const k of Object.keys(value)) if (!ROUTE_PIN_KEYS.has(k)) throw new AiError(422, 'input.invalid', `route has no field ${k}; send { provider, model }`);++ const provider = typeof value.provider === 'string' ? value.provider.trim() : '';++ if (!provider || provider.length > 128 || !/^[A-Za-z0-9][A-Za-z0-9._:-]*$/.test(provider)) {++ throw new AiError(422, 'input.invalid', 'route.provider must name a provider from GET /api/v1/routes');++ }++ let model = null;++ if (value.model != null) {++ if (typeof value.model !== 'string') throw new AiError(422, 'input.invalid', 'route.model must be a model name or null');++ model = value.model.trim();++ if (!model || model.length > 200) throw new AiError(422, 'input.invalid', 'route.model must be a model name or null');++ }++ return { provider, model };++}++++function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null, tiers = null, projectSecrets = null, pool = null }) {const inflight = new Map(); // run id -> { controller, promise, release, caller }const queue = []; // run ids waiting for the workerlet active = 0;@@@ -160,6 +186,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowconst wf = body.version ? await registry.getWorkflow(wfKey, Number(body.version)) : await registry.activeWorkflow(wfKey);if (!wf) throw new AiError(404, 'workflow.not_found', `no active workflow ${wfKey}${body.version ? ` v${body.version}` : ''}`);if (body.version && !['active', 'deprecated'].includes(wf.status)) throw new AiError(409, 'workflow.inactive', `${wfKey} v${wf.version} is ${wf.status}`);++ const pinned = routePinRequest(body.route, principal, wf);++ if (pinned && body.credential != null) throw new AiError(422, 'input.invalid', 'a pinned route cannot be combined with a person\'s credential; the pin is the provider');validateRefs(body);const input = body.input === undefined ? {} : body.input;const inputJson = stableStringify(input);@@@ -177,7 +205,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (idem) {const prev = await db.prepare('SELECT * FROM runs WHERE requester_type = ? AND requester_id = ? AND idempotency_key = ?').get(requester.type, requester.id, idem);if (prev) {-- if (prev.input_hash !== inputHash || prev.workflow_key !== wf.key) throw new AiError(409, 'idempotency.conflict', 'this idempotency_key was used with a different workflow or input');++ const prevPin = parseJson(prev.options, {}).pinned_offer || null;++ if (prev.input_hash !== inputHash || prev.workflow_key !== wf.key || stableStringify(prevPin) !== stableStringify(pinned)) {++ throw new AiError(409, 'idempotency.conflict', 'this idempotency_key was used with a different workflow, input or route');++ }return { run: decode(prev), created: false, promise: inflight.get(prev.id) ? inflight.get(prev.id).promise : null };}}@@@ -186,6 +217,18 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowconst tpl = llmStep ? await registry.activeTemplate(llmStep.template) : null;const routeKey = (wf.steps[0] && wf.steps[0].route) || (tpl && tpl.default_route) || wf.default_route || null;const route = routeKey && !String(routeKey).includes('{') ? await registry.resolveRoute(routeKey) : null;++ if (pinned) {++ if (!route || route.disabled) throw new AiError(503, 'route.unavailable', `route ${routeKey || '(none)'} is not available, so its offers cannot be listed`);++ if (!pool) throw new AiError(503, 'route.unavailable', 'routing is not available on this service');++ const step = wf.steps.find((x) => x.kind === 'llm');++ const operation = (step && step.operation) || 'generate';++ const offers = await pool.offers(route, operation, { requesterType: requester.type, requesterId: requester.id });++ const offered = offers.some((o) => o.provider === pinned.provider && (pinned.model == null || o.model === pinned.model));++ if (!offered) {++ const shown = pinned.model ? `${pinned.provider}/${pinned.model}` : pinned.provider;++ throw new AiError(422, 'input.invalid', `route ${shown} is not offered for ${operation} to this caller; GET /api/v1/routes?operation=${operation} lists the offers`);++ }++ }const options = body.options && typeof body.options === 'object' ? body.options : {};// A person's own key (WS-O task 2): only the service that stored it, within its daily budget; never cached.const credentialSubject = body.credential && typeof body.credential === 'object' ? String(body.credential.subject || '') : null;@@@ -206,6 +249,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowrequesterType: requester.type, requesterId: requester.id,actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,...(app ? { appRun: true, projectId: app.projectId, sandbox: app.sandbox } : {}),++ ...(pinned ? { routePin: pinned } : {}),// POST /v1/chat/completions with stream: true: the upstream's deltas, as they arrive (kept in memory only)....(typeof onDelta === 'function' ? { onDelta } : {}),};@@@ -221,13 +265,15 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowsource_service: requester.type === 'service' ? requester.id : null,target: body.target ? JSON.stringify(body.target) : null, target_key: ctx.targetKey,input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,-- options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider, debug: Boolean(options.debug), input_retained: kept.full,-- ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(projectProvider ? { project_provider: projectProvider } : {}) }), created_at: now,++ options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider && !pinned, debug: Boolean(options.debug), input_retained: kept.full,++ ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(projectProvider ? { project_provider: projectProvider } : {}), ...(pinned ? { pinned_offer: pinned } : {}) }), created_at: now,};base.input = stableStringify(kept.value);// Cache (scoped; never crosses requester/actor/target/attribution)-- const scope = options.cache === false || credentialSubject !== null || projectProvider ? null : cache.scopeFor(wf.cache_mode, ctx);++ // A pinned run is never served from (or written to) the shared cache: the caller asked for THIS model,++ // and another model's answer under the same prompt would silently break the pin.++ const scope = options.cache === false || credentialSubject !== null || projectProvider || pinned ? null : cache.scopeFor(wf.cache_mode, ctx);let cacheKey = null;if (scope) {cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });@@@ -275,7 +321,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowawait events.runChanged(await getRow(id));});} catch (err) { if (hold) await tiers.release(hold); throw err; }-- await registry.audit(principal.sub, retryOf ? 'run.retry' : 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, target: ctx.targetKey, retry_of: retryOf } });++ await registry.audit(principal.sub, retryOf ? 'run.retry' : 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, target: ctx.targetKey, retry_of: retryOf, ...(pinned ? { pinned_offer: pinned } : {}) } });const controller = new AbortController();let release;@@@ -358,6 +404,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now// limited to free and local capacity when it is a sandbox run or when no tier budget governs its// project (AI_GOVERN_TIERS off): metered capacity is only ever spent under a project budget.appRun: Boolean(ctx.appRun), freeOnly: Boolean(ctx.appRun && (ctx.sandbox || !(tiers && tiers.enabled))),++ ...(ctx.routePin ? { routePin: ctx.routePin } : {}),...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}),...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}) });if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');@@@ -439,11 +486,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (!row) throw new AiError(404, 'run.not_found', 'no such run');assertOwner(row, principal, 'ai.workflow.manage', principalHas);if (!['failed', 'cancelled'].includes(row.status)) throw new AiError(409, 'run.not_retryable', `only failed or cancelled runs can be retried (this one is ${row.status})`);-- if (parseJson(row.options, {}).input_retained === false) throw new AiError(409, 'run.input_not_retained', 'this run did not keep its input (inline media or a passthrough prompt); submit it again');++ const opts = parseJson(row.options, {});++ if (opts.input_retained === false) throw new AiError(409, 'run.input_not_retained', 'this run did not keep its input (inline media or a passthrough prompt); submit it again');return await create({workflow: row.workflow_key, input: parseJson(row.input, {}), target: parseJson(row.target, null) || undefined,attribution: parseJson(row.attribution, null) || undefined, on_behalf_of: parseJson(row.on_behalf_of, null) || undefined,-- options: { ...parseJson(row.options, {}), cache: false },++ route: opts.pinned_offer || undefined,++ options: { ...opts, cache: false },}, principal, { trace, requestId, retryOf: id });}nnew file mode 100644@@@ -0,0 +1,161 @@++'use strict';++// A first-party generate run may pin one offer from GET /api/v1/routes (?operation=generate): the++// run tries that provider/model alone and records it, a model the routing does not offer is 422, an++// app token may not pin, a caller's namespace still fails closed, and quotas apply exactly as before.++const assert = require('assert');++const { boot, request, token, suite, seamServer, ALL } = require('./helpers');++++const t = suite('pin-route');++const admin = token('live', ALL);++// Design's shape: a first-party service with ai.run.create and the ai.* namespace.++const design = token('design', ['ai.run.create', 'ai.run.read'], { ns: ['ai.*'] });++const ULID = '01JAB2C3D4E5F6G7H8J9K0MNPQ';++const app = token('app', ['ai.app.run'], {++ sub: `app:app_${ULID}`, actorType: 'app', ns: [`prj_${ULID}`, `app.p${ULID.toLowerCase()}.*`],++ extra: { project_id: `prj_${ULID}`, env: 'production' },++});++++let h; let a; let b;++const calls = { a: 0, b: 0 };++const seam = (label) => seamServer(() => {++ calls[label]++;++ return { text: `<h1>${label}</h1>`, model: `${label}-model`, usage: { input: 3, output: 4 } };++});++++async function provider(key, url, model) {++ const r = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: {++ key, kind: 'http', base_url: url, auth_mode: 'none', default_model: model,++ capabilities: ['chat', 'generate', 'json'],++ } });++ assert.strictEqual(r.status, 201, r.text);++}++const offersOn = (routes, key) => (routes.find((r) => r.key === key) || {}).offers || [];++++t.test('boot with two offered generate providers and no stub fallback', async () => {++ a = await seam('a');++ b = await seam('b');++ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0' } });++ await provider('seam-a', a.url, 'a-model');++ await provider('seam-b', b.url, 'b-model');++});++++t.test('GET /api/v1/routes?operation=generate lists the offers a pin may name', async () => {++ const r = await request(h.base, 'GET', '/api/v1/routes?operation=generate', { tok: design });++ assert.strictEqual(r.status, 200, r.text);++ const route = r.body.routes.find((x) => x.key === 'default.chat');++ assert.ok(route, 'default.chat is listed');++ const ids = offersOn(r.body.routes, 'default.chat').map((o) => `${o.provider}:${o.model}`).sort();++ assert.deepStrictEqual(ids, ['seam-a:a-model', 'seam-b:b-model']);++ assert.deepStrictEqual((r.body.offers || []).map((o) => `${o.provider}:${o.model}`).sort(), ids, "the generate menu is the default route offers");++ assert.ok(route.offers.every((o) => typeof o.provider === 'string' && 'model' in o), 'an offer is { provider, model }');++ const one = await request(h.base, 'GET', '/api/v1/routes/default.chat?operation=generate', { tok: design });++ assert.strictEqual(one.status, 200, one.text);++ assert.deepStrictEqual(one.body.route.offers.map((o) => `${o.provider}:${o.model}`).sort(), ids);++});++++t.test('a pinned generate run uses that offer alone and records it', async () => {++ const before = { a: calls.a, b: calls.b };++ const r = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: design, body: {++ prompt: 'pin this one', route: { provider: 'seam-a', model: 'a-model' }, options: { cache: false },++ } });++ assert.strictEqual(r.status, 201, r.text);++ assert.strictEqual(r.body.run.provenance.provider, 'seam-a');++ assert.strictEqual(r.body.run.provenance.model, 'a-model');++ assert.strictEqual(r.body.run.provenance.fallback_used, false);++ assert.match(r.body.run.explain.reasons.join(' '), /pinned to seam-a\/a-model/);++ assert.strictEqual(calls.a, before.a + 1);++ assert.strictEqual(calls.b, before.b, 'the other offer was not tried');++ const req = (await request(h.base, 'GET', `/api/v1/runs/${r.body.run.id}`, { tok: design })).body.requests;++ assert.deepStrictEqual(req.filter((x) => x.status !== 'skipped').map((x) => `${x.provider_key}:${x.model_key}`), ['seam-a:a-model']);++});++++t.test('the same pin works on POST /api/v1/runs for ai.generate', async () => {++ const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: design, body: {++ workflow: 'ai.generate', input: { prompt: 'by workflow' }, route: { provider: 'seam-b', model: 'b-model' }, options: { cache: false },++ } });++ assert.strictEqual(r.status, 201, r.text);++ assert.strictEqual(r.body.run.provenance.provider, 'seam-b');++ assert.strictEqual(r.body.run.provenance.model, 'b-model');++});++++t.test('an offer the routing does not list is 422 with words, and no provider is called', async () => {++ const before = calls.a + calls.b;++ for (const route of [++ { provider: 'seam-a', model: 'not-a-model' },++ { provider: 'nope', model: 'a-model' },++ ]) {++ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: design, body: { prompt: 'x', route } });++ assert.strictEqual(r.status, 422, r.text);++ assert.strictEqual(r.body.code, 'input.invalid');++ assert.match(r.body.detail, /not offered/i);++ }++ const withCredential = await request(h.base, 'POST', '/api/v1/runs', { tok: design, body: {++ workflow: 'ai.generate', input: { prompt: 'x' }, route: { provider: 'seam-a', model: 'a-model' },++ credential: { subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ' },++ } });++ assert.strictEqual(withCredential.status, 422);++ assert.match(withCredential.body.detail, /credential/i);++ assert.strictEqual(calls.a + calls.b, before, 'nothing was dispatched');++});++++t.test('route pinning is only for the generate operation', async () => {++ const r = await request(h.base, 'POST', '/api/v1/chat', { tok: design, body: {++ messages: [{ role: 'user', content: 'hi' }], route: { provider: 'seam-a', model: 'a-model' },++ } });++ assert.strictEqual(r.status, 422);++ assert.strictEqual(r.body.code, 'input.invalid');++ assert.match(r.body.detail, /generate operation/i);++});++++t.test('an app token cannot pin a route', async () => {++ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: app, body: {++ prompt: 'x', route: { provider: 'seam-a', model: 'a-model' },++ } });++ assert.strictEqual(r.status, 403, r.text);++ assert.strictEqual(r.body.code, 'capability.denied');++ assert.match(r.body.detail, /app may not pin/i);++});++++t.test("a caller's namespace fails closed before any pin is considered", async () => {++ const liveOnly = token('live', ['ai.run.create'], { ns: ['live.*'] });++ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: liveOnly, body: {++ prompt: 'x', route: { provider: 'seam-a', model: 'a-model' },++ } });++ assert.strictEqual(r.status, 403);++ assert.strictEqual(r.body.code, 'capability.namespace_denied');++});++++t.test('an idempotency replay keeps its pin; a different pin under the same key conflicts', async () => {++ const body = { prompt: 'idem pin', route: { provider: 'seam-a', model: 'a-model' }, idempotency_key: 'pin-idem-0001', options: { cache: false } };++ const first = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: design, body });++ assert.strictEqual(first.status, 201, first.text);++ const replay = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: design, body });++ assert.strictEqual(replay.status, 200, replay.text);++ assert.strictEqual(replay.body.run.id, first.body.run.id);++ assert.strictEqual(replay.body.run.provenance.model, 'a-model');++ const other = await request(h.base, 'POST', '/api/v1/generate', { tok: design, body: { ...body, route: { provider: 'seam-b', model: 'b-model' } } });++ assert.strictEqual(other.status, 409);++ assert.strictEqual(other.body.code, 'idempotency.conflict');++});++++t.test('quotas still apply to a pinned run', async () => {++ const pinner = token('pinner', ['ai.run.create'], { ns: ['ai.*'] });++ const q = await request(h.base, 'POST', '/api/v1/quotas', { tok: admin, body: {++ scope_type: 'service', scope_id: 'pinner', window: 'minute', max_requests: 1,++ } });++ assert.strictEqual(q.status, 201, q.text);++ const body = { prompt: 'quota', route: { provider: 'seam-a', model: 'a-model' }, options: { cache: false } };++ const first = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: pinner, body });++ assert.strictEqual(first.status, 201, first.text);++ const second = await request(h.base, 'POST', '/api/v1/generate', { tok: pinner, body });++ assert.strictEqual(second.status, 429, second.text);++ assert.strictEqual(second.body.code, 'quota.exceeded');++});++++t.test('shutdown', async () => {++ await h.stop();++ await a.close();++ await b.close();++});++++t.run();@@@ -64,6 +64,9 @@ t.test('a successful run rolls up one ok attempt: counts, tokens, cost, a percen});t.test('a failed run counts an error and moves the route EWMA by alpha 0.2', async () => {++ // A localhost failure can take as long as the good sample did; anchor the history so the 0.2 move++ // is observable instead of depending on the two latencies differing by a millisecond.++ await h.db.prepare("UPDATE placement_state SET ewma_latency_ms = 500 WHERE route_key = 'default.chat'").run();const before = await placementRow('default.chat');await route('dead', []);const r = await generate('nobody home');
@@ -108,6 +108,7 @@ idempotent replay):
@@ -108,6 +108,7 @@ idempotent replay):
| `POST /api/v1/{chat,generate,summarize,classify,extract,embed,image}` (no `enrich`), `GET /runs/:id` (the app's own runs) | `ai.app.run` for a developer-app token ([Developer apps](#developer-apps)) |
| `POST /api/v1/{chat,generate,summarize,classify,extract,embed,image}` (no `enrich`), `GET /runs/:id` (the app's own runs) | `ai.app.run` for a developer-app token ([Developer apps](#developer-apps)) |
| `GET /api/v1/runs`, `GET /runs/:id` (+ citations, request-log metadata), `GET /runs/:id/citations` | `ai.run.read` (own runs) |
| `GET /api/v1/runs`, `GET /runs/:id` (+ citations, request-log metadata), `GET /runs/:id/citations` | `ai.run.read` (own runs) |
| `GET /api/v1/workflows\|templates\|routes[/:key]` | `ai.run.create`, `ai.workflow.manage` or `ai.usage.read` |
| `GET /api/v1/workflows\|templates\|routes[/:key]` | `ai.run.create`, `ai.workflow.manage` or `ai.usage.read` |
+| `GET /api/v1/routes?operation=generate` | the same read: every route's `offers` (`{provider, model}`), plus the operation's deduplicated `offers` |
| `POST /api/v1/{templates,workflows,routes}/:key/versions`, `…/versions/:v/status` | `ai.workflow.manage` |
| `POST /api/v1/{templates,workflows,routes}/:key/versions`, `…/versions/:v/status` | `ai.workflow.manage` |
| `POST/PATCH /api/v1/providers…`, `…/disable\|enable\|reset`, `POST/PATCH /api/v1/models…`, `POST /api/v1/quotas`, `DELETE /api/v1/cache` | `ai.provider.manage` |
| `POST/PATCH /api/v1/providers…`, `…/disable\|enable\|reset`, `POST/PATCH /api/v1/models…`, `POST /api/v1/quotas`, `DELETE /api/v1/cache` | `ai.provider.manage` |
| `GET /api/v1/status\|usage\|quotas\|requests\|audit\|cache\|providers\|models` | `ai.usage.read` |
| `GET /api/v1/status\|usage\|quotas\|requests\|audit\|cache\|providers\|models` | `ai.usage.read` |
@@ -340,7 +341,17 @@ provider that is disabled, missing credentials, circuit-open or unable to do the
@@ -340,7 +341,17 @@ provider that is disabled, missing credentials, circuit-open or unable to do the
failure; a fallback that answers is recorded on the run (`fallback_used`), in the request log
failure; a fallback that answers is recorded on the run (`fallback_used`), in the request log
(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`
(`fallback = 1`) and in the audit log. Every run response and `GET /api/v1/runs/:id` carry `explain`
(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded
(objective, reasons, selected, and every candidate with its estimated cost, latency and excluded
-reason). `GET /api/v1/runs/:id` adds `explain.usage_readings` when the run has a Billing reading: per
+reason).
+
+`GET /api/v1/routes?operation=generate` lists each route's eligible offers as `{ provider, model }`.
+A first-party service holding `ai.run.create` may pin one on `POST /api/v1/generate` (and on
+`POST /api/v1/runs` for `ai.generate`) with `"route": { "provider": …, "model": … }`. The run tries
+that offer alone, never a fallback; a provider/model the routing does not list for `generate` is
+`422 input.invalid` in words, and the caller's `ns` is still checked first (a workflow outside it is
+`403 capability.namespace_denied`). Developer-app tokens (`ai.app.run`) cannot pin. Quotas, tier budgets and Billing readings are exactly as for an unpinned
+run, and a pinned run is never served from the shared cache.
+
+`GET /api/v1/runs/:id` adds `explain.usage_readings` when the run has a Billing reading: per
reading its `idempotency_key`, `state` (`queued` until Billing accepts it, `sent`, or `failed` when Billing
reading its `idempotency_key`, `state` (`queued` until Billing accepts it, `sent`, or `failed` when Billing
refused it), `attempts`, `queued_at`, `sent_at`, `rejected_at`, `next_attempt_at`, `last_error` (the
refused it), `attempts`, `queued_at`, `sent_at`, `rejected_at`, `next_attempt_at`, `last_error` (the
outbox's last send error, also while a queued reading is retried) and `free_allowance_used` when some
outbox's last send error, also while a queued reading is retried) and `free_allowance_used` when some
@@ -79,14 +79,73 @@ function adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops = crea
@@ -79,14 +79,73 @@ function adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops = crea
templates: { list: registry.listTemplates, get: registry.getTemplate, create: registry.createTemplateVersion, one: 'template' },
templates: { list: registry.listTemplates, get: registry.getTemplate, create: registry.createTemplateVersion, one: 'template' },
workflows: { list: registry.listWorkflows, get: registry.getWorkflow, create: registry.createWorkflowVersion, one: 'workflow' },
workflows: { list: registry.listWorkflows, get: registry.getWorkflow, create: registry.createWorkflowVersion, one: 'workflow' },
};
};
+
+ /**
+ * The offers one route's routing lists for an operation: the eligible provider/model candidates in the
+ * order placement would try them. GET /api/v1/routes publishes these (and the operation can be asked for
+ * with ?operation=generate); a generate run may pin one, and server/runs.js checks it against this same list.
+ */
+ function operationFor(route, requested) {
+ const own = route && route.capability ? route.capability : null;
+ if (!requested) return own || 'chat';
+ if (!own || own === requested) return requested;
+ // A chat route serves generate too; every other route keeps its own operation's offers.
+ return own === 'chat' && requested === 'generate' ? requested : own;
+ }
+ async function routeOffers(route, operation, req) {
+ if (!route || route.status !== 'active') return [];
+ try {
+ let target = route;
+ if (route.alias_of) {
+ target = await registry.resolveRoute(route.key);
+ if (!target || target.disabled) return [];
+ }
+ return await pool.offers(target, operationFor(target, operation), { requesterType: req.principal.subject.type, requesterId: req.principal.subject.id });
+ } catch (err) {
+ log.warn(`[ai] offers for route ${route.key}: ${err.message}`);
+ return [];
+ }
+ }
+ r.get('/api/v1/routes', defRead, wrap(async (req) => {
+ const operation = req.query.operation ? String(req.query.operation) : null;
+ const routes = await registry.listRoutes({ history: req.query.history === '1', namespace: req.query.namespace });
+ for (const route of routes) route.offers = await routeOffers(route, operation, req);
+ // With ?operation=…, also the operation's own menu. For the default operations that is the
+ // default route's list (ai.generate uses default.chat), so every offer there is pinnable for
+ // that operation; an operation with no default route gets the routes' union.
+ if (!operation) return { routes };
+ const baseKey = operation === 'chat' || operation === 'generate' ? 'default.chat'
+ : operation === 'embed' ? 'default.embedding'
+ : operation === 'image' ? 'default.image' : null;
+ const base = baseKey ? routes.find((x) => x.key === baseKey) : null;
+ const offers = [];
+ const seen = new Set();
+ for (const route of (base ? [base] : routes)) for (const o of route.offers) {
+ const id = `${o.provider}|${o.model || ''}`;
+ if (seen.has(id)) continue;
+ seen.add(id);
+ offers.push(o);
+ }
+ return { routes, offers };
+ }));
+ r.get('/api/v1/routes/:key', defRead, wrap(async (req) => {
+ const latest = await registry.getRoute(req.params.key, req.query.version ? Number(req.query.version) : undefined);
+ if (!latest) throw new AiError(404, 'ai.not_found', `no route ${req.params.key}`);
+ const versions = (await registry.listRoutes({ history: true })).filter(x => x.key === req.params.key).map(x => ({ version: x.version, status: x.status, created_by: x.created_by, created_at: x.created_at }));
+ const operation = req.query.operation ? String(req.query.operation) : null;
+ return { route: { ...latest, offers: await routeOffers(latest, operation, req) }, versions };
+ }));
+
for (const [plural, k] of Object.entries(kinds)) {
for (const [plural, k] of Object.entries(kinds)) {
- r.get(`/api/v1/${plural}`, defRead, wrap(async (req) => ({ [plural]: await k.list({ history: req.query.history === '1', namespace: req.query.namespace }) })));
- r.get(`/api/v1/${plural}/:key`, defRead, wrap(async (req) => {
- const latest = await k.get(req.params.key, req.query.version ? Number(req.query.version) : undefined);
- if (!latest) throw new AiError(404, 'ai.not_found', `no ${k.one} ${req.params.key}`);
- const versions = (await k.list({ history: true })).filter(x => x.key === req.params.key).map(x => ({ version: x.version, status: x.status, created_by: x.created_by, created_at: x.created_at }));
- return { [k.one]: latest, versions };
- }));
+ if (plural !== 'routes') {
+ r.get(`/api/v1/${plural}`, defRead, wrap(async (req) => ({ [plural]: await k.list({ history: req.query.history === '1', namespace: req.query.namespace }) })));
+ r.get(`/api/v1/${plural}/:key`, defRead, wrap(async (req) => {
+ const latest = await k.get(req.params.key, req.query.version ? Number(req.query.version) : undefined);
+ if (!latest) throw new AiError(404, 'ai.not_found', `no ${k.one} ${req.params.key}`);
+ const versions = (await k.list({ history: true })).filter(x => x.key === req.params.key).map(x => ({ version: x.version, status: x.status, created_by: x.created_by, created_at: x.created_at }));
+ return { [k.one]: latest, versions };
+ }));
+ }
// An edit is always a new version; the previous one stays readable.
// An edit is always a new version; the previous one stays readable.
r.post(`/api/v1/${plural}/:key/versions`, defWrite, wrap(async (req, res) => {
r.post(`/api/v1/${plural}/:key/versions`, defWrite, wrap(async (req, res) => {
try {
try {
@@ -16,6 +16,8 @@
@@ -16,6 +16,8 @@
* POST /api/v1/runs/:id/citations attach sources/citations to a finished run (ai.run.create, owner)
* POST /api/v1/runs/:id/citations attach sources/citations to a finished run (ai.run.create, owner)
* POST /api/v1/{chat,generate,summarize,classify,extract,enrich,embed}
* POST /api/v1/{chat,generate,summarize,classify,extract,enrich,embed}
* direct operations: a run of workflow ai.<op>, waited for
* direct operations: a run of workflow ai.<op>, waited for
+ * generate also takes route: {provider, model}, one offer from
+ * GET /api/v1/routes?operation=generate (first-party services only)
*
*
* Developer apps (ADR-014, ai.app.run): POST /api/v1/{chat,generate,summarize,classify,extract,embed}
* Developer apps (ADR-014, ai.app.run): POST /api/v1/{chat,generate,summarize,classify,extract,embed}
* and GET /api/v1/runs/:id also take an app token; the run is the app's project's (see auth.appOrCap,
* and GET /api/v1/runs/:id also take an app token; the run is the app's project's (see auth.appOrCap,
@@ -174,9 +176,9 @@ function runsRouter({ runs, registry, auth, config, log = console }) {
@@ -174,9 +176,9 @@ function runsRouter({ runs, registry, auth, config, log = console }) {
r.post(`/api/v1/${op}`, DIRECT_OPS.includes(`ai.${op}`) ? createApp : create, async (req, res) => {
r.post(`/api/v1/${op}`, DIRECT_OPS.includes(`ai.${op}`) ? createApp : create, async (req, res) => {
try {
try {
const body = req.body || {};
const body = req.body || {};
- const { idempotency_key: idem, target, attribution, on_behalf_of: obo, options, ...input } = body;
+ const { idempotency_key: idem, target, attribution, on_behalf_of: obo, options, route, ...input } = body;
const wait = req.query.wait != null ? Math.max(0, Math.min(Number(req.query.wait) || 0, config.runs.maxWaitMs)) : config.runs.maxWaitMs;
const wait = req.query.wait != null ? Math.max(0, Math.min(Number(req.query.wait) || 0, config.runs.maxWaitMs)) : config.runs.maxWaitMs;
- await createAndRespond(req, res, { workflow: `ai.${op}`, input, idempotency_key: idem, target, attribution, on_behalf_of: obo, options }, wait);
+ await createAndRespond(req, res, { workflow: `ai.${op}`, input, idempotency_key: idem, target, attribution, on_behalf_of: obo, options, route }, wait);
} catch (err) { sendError(res, err, req.ov, log); }
} catch (err) { sendError(res, err, req.ov, log); }
});
});
}
}
@@ -55,7 +55,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
@@ -55,7 +55,7 @@ async function start({ config, db: givenDb = null, clock = { now: () => Date.now
// A person's own provider keys (WS-O task 2): stored by the service holding their consent, used by their runs only.
// A person's own provider keys (WS-O task 2): stored by the service holding their consent, used by their runs only.
const credentials = require('./credentials').createCredentials({ db, config, clock });
const credentials = require('./credentials').createCredentials({ db, config, clock });
const projectSecrets = require('./project-secrets').createProjectSecrets({ config, env, fetchImpl });
const projectSecrets = require('./project-secrets').createProjectSecrets({ config, env, fetchImpl });
- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, tiers, projectSecrets });
+ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials, tiers, projectSecrets, pool });
await seed({ registry, quotas, config, env, db });
await seed({ registry, quotas, config, env, db });
const interrupted = await runs.recoverInterrupted();
const interrupted = await runs.recoverInterrupted();
if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);
if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);
@@ -449,6 +449,25 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -449,6 +449,25 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
throw new AiError(502, 'credential.provider_failed', `the person's own provider did not answer: ${String(lastErr && lastErr.message || lastErr).slice(0, 200)}`);
throw new AiError(502, 'credential.provider_failed', `the person's own provider did not answer: ${String(lastErr && lastErr.message || lastErr).slice(0, 200)}`);
}
}
+ /**
+ * The offers the routing lists for one operation of one route: the eligible candidates in the
+ * order placement would try them, as { provider, model }. GET /api/v1/routes publishes these and
+ * a pinned run is accepted only when it names one of them.
+ */
+ async function offers(route, operation, ctx = {}) {
+ const { explain } = await candidates(route, [operation], operation, ctx);
+ const out = [];
+ const seen = new Set();
+ for (const c of explain.candidates) {
+ if (!c || !c.eligible || !c.provider) continue;
+ const id = offerId(c.provider, c.model || null);
+ if (seen.has(id)) continue;
+ seen.add(id);
+ out.push({ provider: c.provider, model: c.model || null });
+ }
+ return out;
+ }
+
async function execute(route, operation, req, ctx = {}) {
async function execute(route, operation, req, ctx = {}) {
if (ctx.credential) return await executeWithCredential(ctx.credential, operation, req, ctx);
if (ctx.credential) return await executeWithCredential(ctx.credential, operation, req, ctx);
const features = [operation];
const features = [operation];
@@ -461,7 +480,19 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -461,7 +480,19 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
// after that is final, never retried or handed to a fallback whose words would follow the first one's.
// after that is final, never retried or handed to a fallback whose words would follow the first one's.
let streamed = false;
let streamed = false;
const onDelta = typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
const onDelta = typeof req.onDelta === 'function' ? (d) => { streamed = true; req.onDelta(d); } : null;
- const { order: list, explain } = await candidates(route, features, operation, ctx);
+ let { order: list, explain } = await candidates(route, features, operation, ctx);
+ // A first-party service's generate run may name one offer from GET /api/v1/routes. The run then
+ // tries that offer alone: no fallback to another model, so the caller gets the model it asked for
+ // or an explicit failure. server/runs.js validates the pin against this same offer list first.
+ if (ctx.routePin && operation === 'generate') {
+ const pin = ctx.routePin;
+ const chosen = list.find((c) => c.provider === pin.provider && (pin.model == null || c.model === pin.model));
+ if (!chosen) {
+ throw new AiError(422, 'route.not_offered', `route ${pin.model ? `${pin.provider}/${pin.model}` : pin.provider} is not offered for generate; GET /api/v1/routes?operation=generate lists the offers`);
+ }
+ list = [chosen];
+ explain.reasons = [`pinned to ${chosen.model ? `${chosen.provider}/${chosen.model}` : chosen.provider} for this run`, ...(explain.reasons || [])];
+ }
for (let i = 0; i < list.length; i++) {
for (let i = 0; i < list.length; i++) {
const c = list[i];
const c = list[i];
const entry = await adapter(c.provider);
const entry = await adapter(c.provider);
@@ -548,7 +579,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -548,7 +579,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried, explain });
throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried, explain });
}
}
- return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, rateCardsFor, providerStates, stats, statFor, candidates };
+ return { adapter, health, resetHealth, execute, executeWithCredential, offers, priceFor, costOf, rateCardsFor, providerStates, stats, statFor, candidates };
}
}
module.exports = { createProviderPool };
module.exports = { createProviderPool };
@@ -73,7 +73,33 @@ function projectProviderOf(spec) {
@@ -73,7 +73,33 @@ function projectProviderOf(spec) {
return { base_url: `${u.origin}${u.pathname.replace(/\/+$/, '')}`, model: String(spec.model), secret: String(spec.secret) };
return { base_url: `${u.origin}${u.pathname.replace(/\/+$/, '')}`, model: String(spec.model), secret: String(spec.secret) };
}
}
-function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null, tiers = null, projectSecrets = null }) {
+/**
+ * A generate run's `route: { provider, model }`: one offer from GET /api/v1/routes. Only a first-party
+ * service may ask for one, and only on the generate operation; the offer itself is checked against the
+ * routing's own list later, once the run's route is resolved.
+ */
+const ROUTE_PIN_KEYS = new Set(['provider', 'model']);
+function routePinRequest(value, principal, wf) {
+ if (value == null) return null;
+ if (principal.kind === 'app') throw new AiError(403, 'capability.denied', 'a developer app may not pin a route; use the routing pool');
+ if (!principal.subject || principal.subject.type !== 'service') throw new AiError(403, 'capability.denied', 'only a first-party service may pin a route');
+ if (!wf || wf.key !== 'ai.generate') throw new AiError(422, 'input.invalid', 'route pinning is available on the generate operation (ai.generate) only');
+ if (typeof value !== 'object' || Array.isArray(value)) throw new AiError(422, 'input.invalid', 'route must be { provider, model }');
+ for (const k of Object.keys(value)) if (!ROUTE_PIN_KEYS.has(k)) throw new AiError(422, 'input.invalid', `route has no field ${k}; send { provider, model }`);
+ const provider = typeof value.provider === 'string' ? value.provider.trim() : '';
+ if (!provider || provider.length > 128 || !/^[A-Za-z0-9][A-Za-z0-9._:-]*$/.test(provider)) {
+ throw new AiError(422, 'input.invalid', 'route.provider must name a provider from GET /api/v1/routes');
+ }
+ let model = null;
+ if (value.model != null) {
+ if (typeof value.model !== 'string') throw new AiError(422, 'input.invalid', 'route.model must be a model name or null');
+ model = value.model.trim();
+ if (!model || model.length > 200) throw new AiError(422, 'input.invalid', 'route.model must be a model name or null');
+ }
+ return { provider, model };
+}
+
+function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null, tiers = null, projectSecrets = null, pool = null }) {
const inflight = new Map(); // run id -> { controller, promise, release, caller }
const inflight = new Map(); // run id -> { controller, promise, release, caller }
const queue = []; // run ids waiting for the worker
const queue = []; // run ids waiting for the worker
let active = 0;
let active = 0;
@@ -160,6 +186,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -160,6 +186,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const wf = body.version ? await registry.getWorkflow(wfKey, Number(body.version)) : await registry.activeWorkflow(wfKey);
const wf = body.version ? await registry.getWorkflow(wfKey, Number(body.version)) : await registry.activeWorkflow(wfKey);
if (!wf) throw new AiError(404, 'workflow.not_found', `no active workflow ${wfKey}${body.version ? ` v${body.version}` : ''}`);
if (!wf) throw new AiError(404, 'workflow.not_found', `no active workflow ${wfKey}${body.version ? ` v${body.version}` : ''}`);
if (body.version && !['active', 'deprecated'].includes(wf.status)) throw new AiError(409, 'workflow.inactive', `${wfKey} v${wf.version} is ${wf.status}`);
if (body.version && !['active', 'deprecated'].includes(wf.status)) throw new AiError(409, 'workflow.inactive', `${wfKey} v${wf.version} is ${wf.status}`);
+ const pinned = routePinRequest(body.route, principal, wf);
+ if (pinned && body.credential != null) throw new AiError(422, 'input.invalid', 'a pinned route cannot be combined with a person\'s credential; the pin is the provider');
validateRefs(body);
validateRefs(body);
const input = body.input === undefined ? {} : body.input;
const input = body.input === undefined ? {} : body.input;
const inputJson = stableStringify(input);
const inputJson = stableStringify(input);
@@ -177,7 +205,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -177,7 +205,10 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (idem) {
if (idem) {
const prev = await db.prepare('SELECT * FROM runs WHERE requester_type = ? AND requester_id = ? AND idempotency_key = ?').get(requester.type, requester.id, idem);
const prev = await db.prepare('SELECT * FROM runs WHERE requester_type = ? AND requester_id = ? AND idempotency_key = ?').get(requester.type, requester.id, idem);
if (prev) {
if (prev) {
- if (prev.input_hash !== inputHash || prev.workflow_key !== wf.key) throw new AiError(409, 'idempotency.conflict', 'this idempotency_key was used with a different workflow or input');
+ const prevPin = parseJson(prev.options, {}).pinned_offer || null;
+ if (prev.input_hash !== inputHash || prev.workflow_key !== wf.key || stableStringify(prevPin) !== stableStringify(pinned)) {
+ throw new AiError(409, 'idempotency.conflict', 'this idempotency_key was used with a different workflow, input or route');
+ }
return { run: decode(prev), created: false, promise: inflight.get(prev.id) ? inflight.get(prev.id).promise : null };
return { run: decode(prev), created: false, promise: inflight.get(prev.id) ? inflight.get(prev.id).promise : null };
}
}
}
}
@@ -186,6 +217,18 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -186,6 +217,18 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const tpl = llmStep ? await registry.activeTemplate(llmStep.template) : null;
const tpl = llmStep ? await registry.activeTemplate(llmStep.template) : null;
const routeKey = (wf.steps[0] && wf.steps[0].route) || (tpl && tpl.default_route) || wf.default_route || null;
const routeKey = (wf.steps[0] && wf.steps[0].route) || (tpl && tpl.default_route) || wf.default_route || null;
const route = routeKey && !String(routeKey).includes('{') ? await registry.resolveRoute(routeKey) : null;
const route = routeKey && !String(routeKey).includes('{') ? await registry.resolveRoute(routeKey) : null;
+ if (pinned) {
+ if (!route || route.disabled) throw new AiError(503, 'route.unavailable', `route ${routeKey || '(none)'} is not available, so its offers cannot be listed`);
+ if (!pool) throw new AiError(503, 'route.unavailable', 'routing is not available on this service');
+ const step = wf.steps.find((x) => x.kind === 'llm');
+ const operation = (step && step.operation) || 'generate';
+ const offers = await pool.offers(route, operation, { requesterType: requester.type, requesterId: requester.id });
+ const offered = offers.some((o) => o.provider === pinned.provider && (pinned.model == null || o.model === pinned.model));
+ if (!offered) {
+ const shown = pinned.model ? `${pinned.provider}/${pinned.model}` : pinned.provider;
+ throw new AiError(422, 'input.invalid', `route ${shown} is not offered for ${operation} to this caller; GET /api/v1/routes?operation=${operation} lists the offers`);
+ }
+ }
const options = body.options && typeof body.options === 'object' ? body.options : {};
const options = body.options && typeof body.options === 'object' ? body.options : {};
// A person's own key (WS-O task 2): only the service that stored it, within its daily budget; never cached.
// A person's own key (WS-O task 2): only the service that stored it, within its daily budget; never cached.
const credentialSubject = body.credential && typeof body.credential === 'object' ? String(body.credential.subject || '') : null;
const credentialSubject = body.credential && typeof body.credential === 'object' ? String(body.credential.subject || '') : null;
@@ -206,6 +249,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -206,6 +249,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
requesterType: requester.type, requesterId: requester.id,
requesterType: requester.type, requesterId: requester.id,
actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,
actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,
...(app ? { appRun: true, projectId: app.projectId, sandbox: app.sandbox } : {}),
...(app ? { appRun: true, projectId: app.projectId, sandbox: app.sandbox } : {}),
+ ...(pinned ? { routePin: pinned } : {}),
// POST /v1/chat/completions with stream: true: the upstream's deltas, as they arrive (kept in memory only).
// POST /v1/chat/completions with stream: true: the upstream's deltas, as they arrive (kept in memory only).
...(typeof onDelta === 'function' ? { onDelta } : {}),
...(typeof onDelta === 'function' ? { onDelta } : {}),
};
};
@@ -221,13 +265,15 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -221,13 +265,15 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
source_service: requester.type === 'service' ? requester.id : null,
source_service: requester.type === 'service' ? requester.id : null,
target: body.target ? JSON.stringify(body.target) : null, target_key: ctx.targetKey,
target: body.target ? JSON.stringify(body.target) : null, target_key: ctx.targetKey,
input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,
input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,
- options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider, debug: Boolean(options.debug), input_retained: kept.full,
- ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(projectProvider ? { project_provider: projectProvider } : {}) }), created_at: now,
+ options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null && !projectProvider && !pinned, debug: Boolean(options.debug), input_retained: kept.full,
+ ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}), ...(projectProvider ? { project_provider: projectProvider } : {}), ...(pinned ? { pinned_offer: pinned } : {}) }), created_at: now,
};
};
base.input = stableStringify(kept.value);
base.input = stableStringify(kept.value);
// Cache (scoped; never crosses requester/actor/target/attribution)
// Cache (scoped; never crosses requester/actor/target/attribution)
- const scope = options.cache === false || credentialSubject !== null || projectProvider ? null : cache.scopeFor(wf.cache_mode, ctx);
+ // A pinned run is never served from (or written to) the shared cache: the caller asked for THIS model,
+ // and another model's answer under the same prompt would silently break the pin.
+ const scope = options.cache === false || credentialSubject !== null || projectProvider || pinned ? null : cache.scopeFor(wf.cache_mode, ctx);
let cacheKey = null;
let cacheKey = null;
if (scope) {
if (scope) {
cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });
cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });
@@ -275,7 +321,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -275,7 +321,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
await events.runChanged(await getRow(id));
await events.runChanged(await getRow(id));
});
});
} catch (err) { if (hold) await tiers.release(hold); throw err; }
} catch (err) { if (hold) await tiers.release(hold); throw err; }
- await registry.audit(principal.sub, retryOf ? 'run.retry' : 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, target: ctx.targetKey, retry_of: retryOf } });
+ await registry.audit(principal.sub, retryOf ? 'run.retry' : 'run.create', 'run', id, { trace, metadata: { workflow: wf.key, version: wf.version, target: ctx.targetKey, retry_of: retryOf, ...(pinned ? { pinned_offer: pinned } : {}) } });
const controller = new AbortController();
const controller = new AbortController();
let release;
let release;
@@ -358,6 +404,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -358,6 +404,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// limited to free and local capacity when it is a sandbox run or when no tier budget governs its
// limited to free and local capacity when it is a sandbox run or when no tier budget governs its
// project (AI_GOVERN_TIERS off): metered capacity is only ever spent under a project budget.
// project (AI_GOVERN_TIERS off): metered capacity is only ever spent under a project budget.
appRun: Boolean(ctx.appRun), freeOnly: Boolean(ctx.appRun && (ctx.sandbox || !(tiers && tiers.enabled))),
appRun: Boolean(ctx.appRun), freeOnly: Boolean(ctx.appRun && (ctx.sandbox || !(tiers && tiers.enabled))),
+ ...(ctx.routePin ? { routePin: ctx.routePin } : {}),
...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}),
...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}),
...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}) });
...(typeof ctx.onDelta === 'function' ? { onDelta: ctx.onDelta } : {}) });
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
@@ -439,11 +486,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -439,11 +486,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (!row) throw new AiError(404, 'run.not_found', 'no such run');
if (!row) throw new AiError(404, 'run.not_found', 'no such run');
assertOwner(row, principal, 'ai.workflow.manage', principalHas);
assertOwner(row, principal, 'ai.workflow.manage', principalHas);
if (!['failed', 'cancelled'].includes(row.status)) throw new AiError(409, 'run.not_retryable', `only failed or cancelled runs can be retried (this one is ${row.status})`);
if (!['failed', 'cancelled'].includes(row.status)) throw new AiError(409, 'run.not_retryable', `only failed or cancelled runs can be retried (this one is ${row.status})`);
- if (parseJson(row.options, {}).input_retained === false) throw new AiError(409, 'run.input_not_retained', 'this run did not keep its input (inline media or a passthrough prompt); submit it again');
+ const opts = parseJson(row.options, {});
+ if (opts.input_retained === false) throw new AiError(409, 'run.input_not_retained', 'this run did not keep its input (inline media or a passthrough prompt); submit it again');
return await create({
return await create({
workflow: row.workflow_key, input: parseJson(row.input, {}), target: parseJson(row.target, null) || undefined,
workflow: row.workflow_key, input: parseJson(row.input, {}), target: parseJson(row.target, null) || undefined,
attribution: parseJson(row.attribution, null) || undefined, on_behalf_of: parseJson(row.on_behalf_of, null) || undefined,
attribution: parseJson(row.attribution, null) || undefined, on_behalf_of: parseJson(row.on_behalf_of, null) || undefined,
- options: { ...parseJson(row.options, {}), cache: false },
+ route: opts.pinned_offer || undefined,
+ options: { ...opts, cache: false },
}, principal, { trace, requestId, retryOf: id });
}, principal, { trace, requestId, retryOf: id });
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,161 @@
@@ -0,0 +1,161 @@
+'use strict';
+// A first-party generate run may pin one offer from GET /api/v1/routes (?operation=generate): the
+// run tries that provider/model alone and records it, a model the routing does not offer is 422, an
+// app token may not pin, a caller's namespace still fails closed, and quotas apply exactly as before.
+const assert = require('assert');
+const { boot, request, token, suite, seamServer, ALL } = require('./helpers');
+
+const t = suite('pin-route');
+const admin = token('live', ALL);
+// Design's shape: a first-party service with ai.run.create and the ai.* namespace.
+const design = token('design', ['ai.run.create', 'ai.run.read'], { ns: ['ai.*'] });
+const ULID = '01JAB2C3D4E5F6G7H8J9K0MNPQ';
+const app = token('app', ['ai.app.run'], {
+ sub: `app:app_${ULID}`, actorType: 'app', ns: [`prj_${ULID}`, `app.p${ULID.toLowerCase()}.*`],
+ extra: { project_id: `prj_${ULID}`, env: 'production' },
+});
+
+let h; let a; let b;
+const calls = { a: 0, b: 0 };
+const seam = (label) => seamServer(() => {
+ calls[label]++;
+ return { text: `<h1>${label}</h1>`, model: `${label}-model`, usage: { input: 3, output: 4 } };
+});
+
+async function provider(key, url, model) {
+ const r = await request(h.base, 'POST', '/api/v1/providers', { tok: admin, body: {
+ key, kind: 'http', base_url: url, auth_mode: 'none', default_model: model,
+ capabilities: ['chat', 'generate', 'json'],
+ } });
+ assert.strictEqual(r.status, 201, r.text);
+}
+const offersOn = (routes, key) => (routes.find((r) => r.key === key) || {}).offers || [];
+
+t.test('boot with two offered generate providers and no stub fallback', async () => {
+ a = await seam('a');
+ b = await seam('b');
+ h = await boot({ env: { AI_STUB_FALLBACK: 'false', AI_PROVIDER_RETRIES: '0' } });
+ await provider('seam-a', a.url, 'a-model');
+ await provider('seam-b', b.url, 'b-model');
+});
+
+t.test('GET /api/v1/routes?operation=generate lists the offers a pin may name', async () => {
+ const r = await request(h.base, 'GET', '/api/v1/routes?operation=generate', { tok: design });
+ assert.strictEqual(r.status, 200, r.text);
+ const route = r.body.routes.find((x) => x.key === 'default.chat');
+ assert.ok(route, 'default.chat is listed');
+ const ids = offersOn(r.body.routes, 'default.chat').map((o) => `${o.provider}:${o.model}`).sort();
+ assert.deepStrictEqual(ids, ['seam-a:a-model', 'seam-b:b-model']);
+ assert.deepStrictEqual((r.body.offers || []).map((o) => `${o.provider}:${o.model}`).sort(), ids, "the generate menu is the default route offers");
+ assert.ok(route.offers.every((o) => typeof o.provider === 'string' && 'model' in o), 'an offer is { provider, model }');
+ const one = await request(h.base, 'GET', '/api/v1/routes/default.chat?operation=generate', { tok: design });
+ assert.strictEqual(one.status, 200, one.text);
+ assert.deepStrictEqual(one.body.route.offers.map((o) => `${o.provider}:${o.model}`).sort(), ids);
+});
+
+t.test('a pinned generate run uses that offer alone and records it', async () => {
+ const before = { a: calls.a, b: calls.b };
+ const r = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: design, body: {
+ prompt: 'pin this one', route: { provider: 'seam-a', model: 'a-model' }, options: { cache: false },
+ } });
+ assert.strictEqual(r.status, 201, r.text);
+ assert.strictEqual(r.body.run.provenance.provider, 'seam-a');
+ assert.strictEqual(r.body.run.provenance.model, 'a-model');
+ assert.strictEqual(r.body.run.provenance.fallback_used, false);
+ assert.match(r.body.run.explain.reasons.join(' '), /pinned to seam-a\/a-model/);
+ assert.strictEqual(calls.a, before.a + 1);
+ assert.strictEqual(calls.b, before.b, 'the other offer was not tried');
+ const req = (await request(h.base, 'GET', `/api/v1/runs/${r.body.run.id}`, { tok: design })).body.requests;
+ assert.deepStrictEqual(req.filter((x) => x.status !== 'skipped').map((x) => `${x.provider_key}:${x.model_key}`), ['seam-a:a-model']);
+});
+
+t.test('the same pin works on POST /api/v1/runs for ai.generate', async () => {
+ const r = await request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok: design, body: {
+ workflow: 'ai.generate', input: { prompt: 'by workflow' }, route: { provider: 'seam-b', model: 'b-model' }, options: { cache: false },
+ } });
+ assert.strictEqual(r.status, 201, r.text);
+ assert.strictEqual(r.body.run.provenance.provider, 'seam-b');
+ assert.strictEqual(r.body.run.provenance.model, 'b-model');
+});
+
+t.test('an offer the routing does not list is 422 with words, and no provider is called', async () => {
+ const before = calls.a + calls.b;
+ for (const route of [
+ { provider: 'seam-a', model: 'not-a-model' },
+ { provider: 'nope', model: 'a-model' },
+ ]) {
+ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: design, body: { prompt: 'x', route } });
+ assert.strictEqual(r.status, 422, r.text);
+ assert.strictEqual(r.body.code, 'input.invalid');
+ assert.match(r.body.detail, /not offered/i);
+ }
+ const withCredential = await request(h.base, 'POST', '/api/v1/runs', { tok: design, body: {
+ workflow: 'ai.generate', input: { prompt: 'x' }, route: { provider: 'seam-a', model: 'a-model' },
+ credential: { subject: 'usr_01JAB2C3D4E5F6G7H8J9K0MNPQ' },
+ } });
+ assert.strictEqual(withCredential.status, 422);
+ assert.match(withCredential.body.detail, /credential/i);
+ assert.strictEqual(calls.a + calls.b, before, 'nothing was dispatched');
+});
+
+t.test('route pinning is only for the generate operation', async () => {
+ const r = await request(h.base, 'POST', '/api/v1/chat', { tok: design, body: {
+ messages: [{ role: 'user', content: 'hi' }], route: { provider: 'seam-a', model: 'a-model' },
+ } });
+ assert.strictEqual(r.status, 422);
+ assert.strictEqual(r.body.code, 'input.invalid');
+ assert.match(r.body.detail, /generate operation/i);
+});
+
+t.test('an app token cannot pin a route', async () => {
+ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: app, body: {
+ prompt: 'x', route: { provider: 'seam-a', model: 'a-model' },
+ } });
+ assert.strictEqual(r.status, 403, r.text);
+ assert.strictEqual(r.body.code, 'capability.denied');
+ assert.match(r.body.detail, /app may not pin/i);
+});
+
+t.test("a caller's namespace fails closed before any pin is considered", async () => {
+ const liveOnly = token('live', ['ai.run.create'], { ns: ['live.*'] });
+ const r = await request(h.base, 'POST', '/api/v1/generate', { tok: liveOnly, body: {
+ prompt: 'x', route: { provider: 'seam-a', model: 'a-model' },
+ } });
+ assert.strictEqual(r.status, 403);
+ assert.strictEqual(r.body.code, 'capability.namespace_denied');
+});
+
+t.test('an idempotency replay keeps its pin; a different pin under the same key conflicts', async () => {
+ const body = { prompt: 'idem pin', route: { provider: 'seam-a', model: 'a-model' }, idempotency_key: 'pin-idem-0001', options: { cache: false } };
+ const first = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: design, body });
+ assert.strictEqual(first.status, 201, first.text);
+ const replay = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: design, body });
+ assert.strictEqual(replay.status, 200, replay.text);
+ assert.strictEqual(replay.body.run.id, first.body.run.id);
+ assert.strictEqual(replay.body.run.provenance.model, 'a-model');
+ const other = await request(h.base, 'POST', '/api/v1/generate', { tok: design, body: { ...body, route: { provider: 'seam-b', model: 'b-model' } } });
+ assert.strictEqual(other.status, 409);
+ assert.strictEqual(other.body.code, 'idempotency.conflict');
+});
+
+t.test('quotas still apply to a pinned run', async () => {
+ const pinner = token('pinner', ['ai.run.create'], { ns: ['ai.*'] });
+ const q = await request(h.base, 'POST', '/api/v1/quotas', { tok: admin, body: {
+ scope_type: 'service', scope_id: 'pinner', window: 'minute', max_requests: 1,
+ } });
+ assert.strictEqual(q.status, 201, q.text);
+ const body = { prompt: 'quota', route: { provider: 'seam-a', model: 'a-model' }, options: { cache: false } };
+ const first = await request(h.base, 'POST', '/api/v1/generate?wait=5000', { tok: pinner, body });
+ assert.strictEqual(first.status, 201, first.text);
+ const second = await request(h.base, 'POST', '/api/v1/generate', { tok: pinner, body });
+ assert.strictEqual(second.status, 429, second.text);
+ assert.strictEqual(second.body.code, 'quota.exceeded');
+});
+
+t.test('shutdown', async () => {
+ await h.stop();
+ await a.close();
+ await b.close();
+});
+
+t.run();
@@ -64,6 +64,9 @@ t.test('a successful run rolls up one ok attempt: counts, tokens, cost, a percen
@@ -64,6 +64,9 @@ t.test('a successful run rolls up one ok attempt: counts, tokens, cost, a percen
});
});
t.test('a failed run counts an error and moves the route EWMA by alpha 0.2', async () => {
t.test('a failed run counts an error and moves the route EWMA by alpha 0.2', async () => {
+ // A localhost failure can take as long as the good sample did; anchor the history so the 0.2 move
+ // is observable instead of depending on the two latencies differing by a millisecond.
+ await h.db.prepare("UPDATE placement_state SET ewma_latency_ms = 500 WHERE route_key = 'default.chat'").run();
const before = await placementRow('default.chat');
const before = await placementRow('default.chat');
await route('dead', []);
await route('dead', []);
const r = await generate('nobody home');
const r = await generate('nobody home');